refactor: harden sqlite runtime and packaged startup flow

- refine config loading with --config and current-directory .env

- improve sqlite busy handling and runtime DB tuning

- restore telegram payment notification and wallet input flow

- add behavior-based config and callback recovery tests

src/mq/worker.go

@@ -3,6 +3,7 @@ package mq
 import (
 	"errors"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/assimon/luuu/config"
@@ -17,6 +18,16 @@ import (
 
 const batchSize = 100
 
+const sqliteBusyRetryAttempts = 3
+
+type expirableOrder struct {
+	ID             uint64  `gorm:"column:id"`
+	TradeId        string  `gorm:"column:trade_id"`
+	ReceiveAddress string  `gorm:"column:receive_address"`
+	Token          string  `gorm:"column:token"`
+	ActualAmount   float64 `gorm:"column:actual_amount"`
+}
+
 func runOrderExpirationLoop() {
 	runLoop("order_expiration", processExpiredOrders)
 }
@@ -51,13 +62,16 @@ func safeRun(name string, fn func()) {
 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
+		var orders []expirableOrder
+		err := withSQLiteBusyRetry(func() error {
+			return dao.Mdb.Model(&mdb.Orders{}).
+				Select("id", "trade_id", "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
@@ -88,7 +102,12 @@ func processExpiredOrders() {
 
 func dispatchPendingCallbacks() {
 	maxRetry := config.GetOrderNoticeMaxRetry()
-	orders, err := data.GetPendingCallbackOrders(maxRetry, batchSize)
+	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
@@ -99,29 +118,30 @@ func dispatchPendingCallbacks() {
 		if !isCallbackDue(&order, now, maxRetry) {
 			continue
 		}
-		if _, loaded := callbackInflight.LoadOrStore(order.TradeId, struct{}{}); loaded {
+		tradeID := order.TradeId
+		if _, loaded := callbackInflight.LoadOrStore(tradeID, struct{}{}); loaded {
 			continue
 		}
 
 		select {
 		case callbackLimiter <- struct{}{}:
-			go processCallback(order)
+			go processCallback(tradeID)
 		default:
-			callbackInflight.Delete(order.TradeId)
+			callbackInflight.Delete(tradeID)
 			return
 		}
 	}
 }
 
-func processCallback(order mdb.Orders) {
+func processCallback(tradeID string) {
 	defer func() {
 		<-callbackLimiter
-		callbackInflight.Delete(order.TradeId)
+		callbackInflight.Delete(tradeID)
 	}()
 
-	freshOrder, err := data.GetOrderInfoByTradeId(order.TradeId)
+	freshOrder, err := data.GetOrderInfoByTradeId(tradeID)
 	if err != nil {
-		log.Sugar.Errorf("[mq] reload callback order failed, trade_id=%s, err=%v", order.TradeId, err)
+		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 {
@@ -180,7 +200,30 @@ func cleanupExpiredTransactionLocks() {
 	}
 }
 
-func isCallbackDue(order *mdb.Orders, now time.Time, maxRetry int) bool {
+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
 	}

src/mq/worker_test.go

@@ -185,6 +185,78 @@ func TestDispatchPendingCallbacksHonorsBackoffAndPersistsSuccess(t *testing.T) {
 	}
 }
 
+func TestDispatchPendingCallbacksResumesRetryAfterRestart(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) {
+		attempt := atomic.AddInt32(&requestCount, 1)
+		if attempt == 1 {
+			http.Error(w, "retry later", http.StatusInternalServerError)
+			return
+		}
+		_, _ = io.WriteString(w, "ok")
+	}))
+	defer server.Close()
+
+	order := &mdb.Orders{
+		TradeId:            "trade_callback_restart",
+		OrderId:            "order_callback_restart",
+		Amount:             1,
+		Currency:           "CNY",
+		ActualAmount:       1,
+		ReceiveAddress:     "wallet_restart",
+		Token:              "USDT",
+		Status:             mdb.StatusPaySuccess,
+		NotifyUrl:          server.URL,
+		BlockTransactionId: "block_callback_restart",
+		CallbackNum:        0,
+		CallBackConfirm:    mdb.CallBackConfirmNo,
+	}
+	if err := dao.Mdb.Create(order).Error; err != nil {
+		t.Fatalf("create 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.CallBackConfirmNo && current.CallbackNum == 1
+	})
+
+	if got := atomic.LoadInt32(&requestCount); got != 1 {
+		t.Fatalf("first callback request count = %d, want 1", got)
+	}
+
+	callbackLimiter = make(chan struct{}, 1)
+	callbackInflight = sync.Map{}
+
+	if err := dao.Mdb.Model(order).UpdateColumn("updated_at", time.Now().Add(-2*time.Second)).Error; err != nil {
+		t.Fatalf("age callback order for retry: %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 != 2 {
+		t.Fatalf("total callback request count = %d, want 2", got)
+	}
+}
+
 func waitFor(t *testing.T, timeout time.Duration, fn func() bool) {
 	t.Helper()