refactor: replace redis runtime and queues with sqlite-backed scheduler

2026-07-07 18:26:16 +00:00 · 2026-03-29 17:26:36 +08:00
parent d6e7927605
commit fc8d58024e
22 changed files with 1193 additions and 446 deletions
@@ -1,70 +0,0 @@
-package handle
-
-import (
-	"context"
-	"errors"
-
-	"github.com/assimon/luuu/config"
-	"github.com/assimon/luuu/model/data"
-	"github.com/assimon/luuu/model/mdb"
-	"github.com/assimon/luuu/model/response"
-	"github.com/assimon/luuu/util/http_client"
-	"github.com/assimon/luuu/util/json"
-	"github.com/assimon/luuu/util/log"
-	"github.com/assimon/luuu/util/sign"
-	"github.com/hibiken/asynq"
-)
-
-const QueueOrderCallback = "order:callback"
-
-func NewOrderCallbackQueue(order *mdb.Orders) (*asynq.Task, error) {
-	payload, err := json.Cjson.Marshal(order)
-	if err != nil {
-		return nil, err
-	}
-	return asynq.NewTask(QueueOrderCallback, payload,
-		asynq.Retention(config.GetOrderExpirationTimeDuration()),
-	), nil
-}
-
-func OrderCallbackHandle(ctx context.Context, t *asynq.Task) error {
-	var order mdb.Orders
-	err := json.Cjson.Unmarshal(t.Payload(), &order)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		if err := recover(); err != nil {
-			log.Sugar.Error(err)
-		}
-	}()
-	defer func() {
-		data.SaveCallBackOrdersResp(&order)
-	}()
-	client := http_client.GetHttpClient()
-	orderResp := response.OrderNotifyResponse{
-		TradeId:            order.TradeId,
-		OrderId:            order.OrderId,
-		Amount:             order.Amount,
-		ActualAmount:       order.ActualAmount,
-		Token:              order.Token,
-		BlockTransactionId: order.BlockTransactionId,
-		Status:             mdb.StatusPaySuccess,
-	}
-	signature, err := sign.Get(orderResp, config.GetApiAuthToken())
-	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
-	}
-	body := string(resp.Body())
-	if body != "ok" {
-		order.CallBackConfirm = mdb.CallBackConfirmNo
-		return errors.New("not ok")
-	}
-	order.CallBackConfirm = mdb.CallBackConfirmOk
-	return nil
-}
@@ -1,39 +0,0 @@
-package handle
-
-import (
-	"context"
-
-	"github.com/assimon/luuu/config"
-	"github.com/assimon/luuu/model/data"
-	"github.com/assimon/luuu/model/mdb"
-	"github.com/hibiken/asynq"
-)
-
-const QueueOrderExpiration = "order:expiration"
-
-func NewOrderExpirationQueue(tradeId string) (*asynq.Task, error) {
-	return asynq.NewTask(QueueOrderExpiration, []byte(tradeId),
-		asynq.Retention(config.GetOrderExpirationTimeDuration()),
-	), nil
-}
-
-// OrderExpirationHandle 设置订单过期
-func OrderExpirationHandle(ctx context.Context, t *asynq.Task) error {
-	tradeId := string(t.Payload())
-	orderInfo, err := data.GetOrderInfoByTradeId(tradeId)
-	if err != nil {
-		return err
-	}
-	if orderInfo.ID <= 0 || orderInfo.Status != mdb.StatusWaitPay {
-		return nil
-	}
-	err = data.UpdateOrderIsExpirationById(orderInfo.ID)
-	if err != nil {
-		return err
-	}
-	err = data.UnLockTransaction(orderInfo.Token, orderInfo.ActualAmount)
-	if err != nil {
-		return err
-	}
-	return nil
-}
@@ -1,52 +1,22 @@
 package mq

 import (
-	"fmt"
+	"sync"

-	"github.com/assimon/luuu/mq/handle"
-	"github.com/assimon/luuu/util/log"
-	"github.com/hibiken/asynq"
-	"github.com/spf13/viper"
+	"github.com/assimon/luuu/config"
 )

-var MClient *asynq.Client
+var (
+	startOnce        sync.Once
+	callbackLimiter  chan struct{}
+	callbackInflight sync.Map
+)

 func Start() {
-	redis := asynq.RedisClientOpt{
-		Addr: fmt.Sprintf(
-			"%s:%s",
-			viper.GetString("redis_host"),
-			viper.GetString("redis_port")),
-		DB:       viper.GetInt("redis_db"),
-		Password: viper.GetString("redis_passwd"),
-	}
-	initClient(redis)
-	go initListen(redis)
-}
-
-func initClient(redis asynq.RedisClientOpt) {
-	MClient = asynq.NewClient(redis)
-}
-
-func initListen(redis asynq.RedisClientOpt) {
-	srv := asynq.NewServer(
-		redis,
-		asynq.Config{
-			// Specify how many concurrent workers to use
-			Concurrency: viper.GetInt("queue_concurrency"),
-			// Optionally specify multiple queues with different priority.
-			Queues: map[string]int{
-				"critical": viper.GetInt("queue_level_critical"),
-				"default":  viper.GetInt("queue_level_default"),
-				"low":      viper.GetInt("queue_level_low"),
-			},
-			Logger: log.Sugar,
-		},
-	)
-	mux := asynq.NewServeMux()
-	mux.HandleFunc(handle.QueueOrderExpiration, handle.OrderExpirationHandle)
-	mux.HandleFunc(handle.QueueOrderCallback, handle.OrderCallbackHandle)
-	if err := srv.Run(mux); err != nil {
-		log.Sugar.Fatalf("[queue] could not run server: %v", err)
-	}
+	startOnce.Do(func() {
+		callbackLimiter = make(chan struct{}, config.GetQueueConcurrency())
+		go runOrderExpirationLoop()
+		go runOrderCallbackLoop()
+		go runTransactionLockCleanupLoop()
+	})
 }
@@ -0,0 +1,213 @@
+package mq
+
+import (
+	"errors"
+	"net/http"
+	"time"
+
+	"github.com/assimon/luuu/config"
+	"github.com/assimon/luuu/model/dao"
+	"github.com/assimon/luuu/model/data"
+	"github.com/assimon/luuu/model/mdb"
+	"github.com/assimon/luuu/model/response"
+	"github.com/assimon/luuu/util/http_client"
+	"github.com/assimon/luuu/util/log"
+	"github.com/assimon/luuu/util/sign"
+)
+
+const batchSize = 100
+
+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 []mdb.Orders
+		err := dao.Mdb.Model(&mdb.Orders{}).
+			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.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()
+	orders, err := data.GetPendingCallbackOrders(maxRetry, batchSize)
+	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
+		}
+		if _, loaded := callbackInflight.LoadOrStore(order.TradeId, struct{}{}); loaded {
+			continue
+		}
+
+		select {
+		case callbackLimiter <- struct{}{}:
+			go processCallback(order)
+		default:
+			callbackInflight.Delete(order.TradeId)
+			return
+		}
+	}
+}
+
+func processCallback(order mdb.Orders) {
+	defer func() {
+		<-callbackLimiter
+		callbackInflight.Delete(order.TradeId)
+	}()
+
+	freshOrder, err := data.GetOrderInfoByTradeId(order.TradeId)
+	if err != nil {
+		log.Sugar.Errorf("[mq] reload callback order failed, trade_id=%s, err=%v", order.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 {
+	client := http_client.GetHttpClient()
+	orderResp := response.OrderNotifyResponse{
+		TradeId:            order.TradeId,
+		OrderId:            order.OrderId,
+		Amount:             order.Amount,
+		ActualAmount:       order.ActualAmount,
+		Token:              order.Token,
+		BlockTransactionId: order.BlockTransactionId,
+		Status:             mdb.StatusPaySuccess,
+	}
+	signature, err := sign.Get(orderResp, config.GetApiAuthToken())
+	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 string(resp.Body()) != "ok" {
+		return errors.New("not ok")
+	}
+	return nil
+}
+
+func cleanupExpiredTransactionLocks() {
+	if err := data.CleanupExpiredTransactionLocks(); err != nil {
+		log.Sugar.Errorf("[mq] cleanup expired transaction locks failed: %v", err)
+	}
+}
+
+func isCallbackDue(order *mdb.Orders, 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
+}
@@ -0,0 +1,191 @@
+package mq
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/assimon/luuu/internal/testutil"
+	"github.com/assimon/luuu/model/dao"
+	"github.com/assimon/luuu/model/data"
+	"github.com/assimon/luuu/model/mdb"
+)
+
+func TestProcessExpiredOrdersExpiresWaitingOrdersAndReleasesLocks(t *testing.T) {
+	cleanup := testutil.SetupTestDatabases(t)
+	defer cleanup()
+
+	order := &mdb.Orders{
+		TradeId:      "trade_expired",
+		OrderId:      "order_expired",
+		Amount:       1,
+		ActualAmount: 1,
+		Token:        "wallet_1",
+		Status:       mdb.StatusWaitPay,
+		NotifyUrl:    "https://merchant.example/callback",
+	}
+	if err := dao.Mdb.Create(order).Error; err != nil {
+		t.Fatalf("create expired order: %v", err)
+	}
+	if err := dao.Mdb.Model(order).UpdateColumn("created_at", time.Now().Add(-20*time.Minute)).Error; err != nil {
+		t.Fatalf("age expired order: %v", err)
+	}
+	if err := data.LockTransaction(order.Token, order.TradeId, order.ActualAmount, time.Hour); err != nil {
+		t.Fatalf("lock expired order: %v", err)
+	}
+
+	recentOrder := &mdb.Orders{
+		TradeId:      "trade_recent",
+		OrderId:      "order_recent",
+		Amount:       1,
+		ActualAmount: 1.01,
+		Token:        "wallet_1",
+		Status:       mdb.StatusWaitPay,
+		NotifyUrl:    "https://merchant.example/callback",
+	}
+	if err := dao.Mdb.Create(recentOrder).Error; err != nil {
+		t.Fatalf("create recent order: %v", err)
+	}
+	if err := data.LockTransaction(recentOrder.Token, recentOrder.TradeId, recentOrder.ActualAmount, time.Hour); err != nil {
+		t.Fatalf("lock recent order: %v", err)
+	}
+
+	processExpiredOrders()
+
+	expired, err := data.GetOrderInfoByTradeId(order.TradeId)
+	if err != nil {
+		t.Fatalf("reload expired order: %v", err)
+	}
+	if expired.Status != mdb.StatusExpired {
+		t.Fatalf("expired order status = %d, want %d", expired.Status, mdb.StatusExpired)
+	}
+	lockTradeID, err := data.GetTradeIdByWalletAddressAndAmount(order.Token, order.ActualAmount)
+	if err != nil {
+		t.Fatalf("expired order lock lookup: %v", err)
+	}
+	if lockTradeID != "" {
+		t.Fatalf("expired order lock still exists: %s", lockTradeID)
+	}
+
+	recent, err := data.GetOrderInfoByTradeId(recentOrder.TradeId)
+	if err != nil {
+		t.Fatalf("reload recent order: %v", err)
+	}
+	if recent.Status != mdb.StatusWaitPay {
+		t.Fatalf("recent order status = %d, want %d", recent.Status, mdb.StatusWaitPay)
+	}
+	lockTradeID, err = data.GetTradeIdByWalletAddressAndAmount(recentOrder.Token, recentOrder.ActualAmount)
+	if err != nil {
+		t.Fatalf("recent order lock lookup: %v", err)
+	}
+	if lockTradeID != recentOrder.TradeId {
+		t.Fatalf("recent order lock = %s, want %s", lockTradeID, recentOrder.TradeId)
+	}
+}
+
+func TestProcessExpiredOrdersKeepsPaidOrdersPaid(t *testing.T) {
+	cleanup := testutil.SetupTestDatabases(t)
+	defer cleanup()
+
+	order := &mdb.Orders{
+		TradeId:            "trade_paid",
+		OrderId:            "order_paid",
+		Amount:             1,
+		ActualAmount:       1,
+		Token:              "wallet_1",
+		Status:             mdb.StatusPaySuccess,
+		NotifyUrl:          "https://merchant.example/callback",
+		BlockTransactionId: "block_paid",
+		CallBackConfirm:    mdb.CallBackConfirmNo,
+	}
+	if err := dao.Mdb.Create(order).Error; err != nil {
+		t.Fatalf("create paid order: %v", err)
+	}
+	if err := dao.Mdb.Model(order).UpdateColumn("created_at", time.Now().Add(-20*time.Minute)).Error; err != nil {
+		t.Fatalf("age paid order: %v", err)
+	}
+
+	processExpiredOrders()
+
+	current, err := data.GetOrderInfoByTradeId(order.TradeId)
+	if err != nil {
+		t.Fatalf("reload paid order: %v", err)
+	}
+	if current.Status != mdb.StatusPaySuccess {
+		t.Fatalf("paid order status = %d, want %d", current.Status, mdb.StatusPaySuccess)
+	}
+	if current.BlockTransactionId != "block_paid" {
+		t.Fatalf("paid order block transaction id = %s, want block_paid", current.BlockTransactionId)
+	}
+}
+
+func TestDispatchPendingCallbacksHonorsBackoffAndPersistsSuccess(t *testing.T) {
+	cleanup := testutil.SetupTestDatabases(t)
+	defer cleanup()
+
+	callbackLimiter = make(chan struct{}, 1)
+	callbackInflight = sync.Map{}
+
+	var requestCount int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		atomic.AddInt32(&requestCount, 1)
+		_, _ = io.WriteString(w, "ok")
+	}))
+	defer server.Close()
+
+	order := &mdb.Orders{
+		TradeId:            "trade_callback",
+		OrderId:            "order_callback",
+		Amount:             1,
+		ActualAmount:       1,
+		Token:              "wallet_1",
+		Status:             mdb.StatusPaySuccess,
+		NotifyUrl:          server.URL,
+		BlockTransactionId: "block_callback",
+		CallbackNum:        1,
+		CallBackConfirm:    mdb.CallBackConfirmNo,
+	}
+	if err := dao.Mdb.Create(order).Error; err != nil {
+		t.Fatalf("create callback order: %v", err)
+	}
+
+	dispatchPendingCallbacks()
+	time.Sleep(200 * time.Millisecond)
+	if got := atomic.LoadInt32(&requestCount); got != 0 {
+		t.Fatalf("unexpected callback count before backoff elapsed: %d", got)
+	}
+
+	if err := dao.Mdb.Model(order).UpdateColumn("updated_at", time.Now().Add(-2*time.Second)).Error; err != nil {
+		t.Fatalf("age callback order: %v", err)
+	}
+	dispatchPendingCallbacks()
+
+	waitFor(t, 3*time.Second, func() bool {
+		current, err := data.GetOrderInfoByTradeId(order.TradeId)
+		if err != nil || current.ID <= 0 {
+			return false
+		}
+		return current.CallBackConfirm == mdb.CallBackConfirmOk && current.CallbackNum == 2
+	})
+
+	if got := atomic.LoadInt32(&requestCount); got != 1 {
+		t.Fatalf("callback request count = %d, want 1", got)
+	}
+}
+
+func waitFor(t *testing.T, timeout time.Duration, fn func() bool) {
+	t.Helper()
+
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if fn() {
+			return
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatal("condition not satisfied before timeout")
+}