package data import ( "errors" "strings" "time" "github.com/GMWalletApp/epusdt/model/dao" "github.com/GMWalletApp/epusdt/model/mdb" "github.com/GMWalletApp/epusdt/model/request" addressutil "github.com/GMWalletApp/epusdt/util/address" "github.com/dromara/carbon/v2" "github.com/shopspring/decimal" "gorm.io/gorm" "gorm.io/gorm/clause" ) var ErrTransactionLocked = errors.New("transaction amount is already locked") type PendingCallbackOrder struct { TradeId string `gorm:"column:trade_id"` CallbackNum int `gorm:"column:callback_num"` CallBackConfirm int `gorm:"column:callback_confirm"` UpdatedAt carbon.Time `gorm:"column:updated_at"` } func normalizeAmount(amount float64, precision int) (int64, string) { precision = NormalizeAmountPrecision(precision) value := decimal.NewFromFloat(amount).Round(int32(precision)) return value.Shift(int32(precision)).IntPart(), value.StringFixed(int32(precision)) } func normalizeLockAmount(amount float64) (int64, string) { return normalizeAmount(amount, GetAmountPrecision()) } func normalizeLockNetwork(network string) string { return strings.ToLower(strings.TrimSpace(network)) } func normalizeLockAddress(network, address string) string { address = strings.TrimSpace(address) network = normalizeLockNetwork(network) if isEVMNetwork(network) { return strings.ToLower(address) } if network == mdb.NetworkTon { if raw, err := addressutil.TonRawAddressKey(address); err == nil { return raw } } return address } func normalizeLockToken(token string) string { return strings.ToUpper(strings.TrimSpace(token)) } func applyLockAddressFilter(tx *gorm.DB, network, address string) *gorm.DB { network = normalizeLockNetwork(network) address = normalizeLockAddress(network, address) if isEVMNetwork(network) { return tx.Where("lower(address) = ?", address) } return tx.Where("address = ?", address) } func activeLocksForAddress(tx *gorm.DB, network, address, token string, now time.Time) *gorm.DB { query := tx.Model(&mdb.TransactionLock{}). Where("network = ?", normalizeLockNetwork(network)). Where("token = ?", normalizeLockToken(token)). Where("expires_at > ?", now) return applyLockAddressFilter(query, network, address) } func lockMatchesAmount(lock mdb.TransactionLock, amount float64) bool { precision := NormalizeAmountPrecision(lock.AmountPrecision) scaledAmount, _ := normalizeAmount(amount, precision) return lock.AmountScaled == scaledAmount } func lockAmountDecimal(lock mdb.TransactionLock) decimal.Decimal { if strings.TrimSpace(lock.AmountText) != "" { if value, err := decimal.NewFromString(lock.AmountText); err == nil { return value } } precision := NormalizeAmountPrecision(lock.AmountPrecision) return decimal.NewFromInt(lock.AmountScaled).Shift(int32(-precision)) } // GetOrderInfoByOrderId fetches an order by merchant order id. func GetOrderInfoByOrderId(orderId string) (*mdb.Orders, error) { order := new(mdb.Orders) err := dao.Mdb.Model(order).Limit(1).Find(order, "order_id = ?", orderId).Error return order, err } // GetOrderInfoByTradeId fetches an order by epusdt trade id. func GetOrderInfoByTradeId(tradeId string) (*mdb.Orders, error) { order := new(mdb.Orders) err := dao.Mdb.Model(order).Limit(1).Find(order, "trade_id = ?", tradeId).Error return order, err } // CreateOrderWithTransaction creates an order in the active database transaction. func CreateOrderWithTransaction(tx *gorm.DB, order *mdb.Orders) error { return tx.Model(order).Create(order).Error } // GetOrderByBlockIdWithTransaction fetches an order by blockchain tx id. func GetOrderByBlockIdWithTransaction(tx *gorm.DB, blockID string) (*mdb.Orders, error) { order := new(mdb.Orders) err := tx.Model(order).Limit(1).Find(order, "block_transaction_id = ?", blockID).Error return order, err } // GetOrderByBlockTransactionIDs fetches the first order whose stored tx id // matches any equivalent spelling supplied by the caller. func GetOrderByBlockTransactionIDs(blockIDs []string) (*mdb.Orders, error) { order := new(mdb.Orders) seen := make(map[string]struct{}, len(blockIDs)) candidates := make([]string, 0, len(blockIDs)) for _, blockID := range blockIDs { blockID = strings.TrimSpace(blockID) if blockID == "" { continue } if _, ok := seen[blockID]; ok { continue } seen[blockID] = struct{}{} candidates = append(candidates, blockID) } if len(candidates) == 0 { return order, nil } err := dao.Mdb.Model(order).Where("block_transaction_id IN ?", candidates).Limit(1).Find(order).Error return order, err } // GetOrderByBlockTransactionIDsCaseInsensitive fetches the first order whose // stored tx id matches any candidate after ASCII case folding. This is used // only for hex-based chain tx ids; case-sensitive signatures must keep using // GetOrderByBlockTransactionIDs. func GetOrderByBlockTransactionIDsCaseInsensitive(blockIDs []string) (*mdb.Orders, error) { order := new(mdb.Orders) seen := make(map[string]struct{}, len(blockIDs)) candidates := make([]string, 0, len(blockIDs)) for _, blockID := range blockIDs { blockID = strings.ToLower(strings.TrimSpace(blockID)) if blockID == "" { continue } if _, ok := seen[blockID]; ok { continue } seen[blockID] = struct{}{} candidates = append(candidates, blockID) } if len(candidates) == 0 { return order, nil } err := dao.Mdb.Model(order).Where("LOWER(block_transaction_id) IN ?", candidates).Limit(1).Find(order).Error return order, err } // OrderSuccessWithTransaction marks an order as paid only if it is still waiting for payment. func OrderSuccessWithTransaction(tx *gorm.DB, req *request.OrderProcessingRequest) (bool, error) { result := tx.Model(&mdb.Orders{}). Where("trade_id = ?", req.TradeId). Where("status = ?", mdb.StatusWaitPay). Updates(map[string]interface{}{ "block_transaction_id": req.BlockTransactionId, "status": mdb.StatusPaySuccess, "callback_confirm": mdb.CallBackConfirmNo, }) return result.RowsAffected > 0, result.Error } // GetPendingCallbackOrders returns the minimal callback scheduling state. func GetPendingCallbackOrders(maxRetry int, limit int) ([]PendingCallbackOrder, error) { var orders []PendingCallbackOrder query := dao.Mdb.Model(&mdb.Orders{}). Select("trade_id", "callback_num", "callback_confirm", "updated_at"). Where("callback_num <= ?", maxRetry). Where("callback_confirm = ?", mdb.CallBackConfirmNo). Where("status = ?", mdb.StatusPaySuccess). Order("updated_at asc") if limit > 0 { query = query.Limit(limit) } err := query.Find(&orders).Error return orders, err } // SaveCallBackOrdersResp persists a callback attempt result. func SaveCallBackOrdersResp(order *mdb.Orders) error { return dao.Mdb.Model(order). Where("id = ?", order.ID). Where("callback_confirm = ?", mdb.CallBackConfirmNo). Updates(map[string]interface{}{ "callback_num": gorm.Expr("callback_num + ?", 1), "callback_confirm": order.CallBackConfirm, }).Error } // UpdateOrderIsExpirationById expires an order only if it is still pending and already timed out. func UpdateOrderIsExpirationById(id uint64, expirationCutoff time.Time) (bool, error) { result := dao.Mdb.Model(mdb.Orders{}). Where("id = ?", id). Where("status = ?", mdb.StatusWaitPay). Where("created_at <= ?", expirationCutoff). Update("status", mdb.StatusExpired) return result.RowsAffected > 0, result.Error } // CountActiveSubOrders counts sub-orders with status=WaitPay under a parent. func CountActiveSubOrders(parentTradeId string) (int64, error) { var count int64 err := dao.Mdb.Model(&mdb.Orders{}). Where("parent_trade_id = ?", parentTradeId). Where("status = ?", mdb.StatusWaitPay). Count(&count).Error return count, err } // GetSubOrderByTokenNetwork finds an existing active sub-order matching token+network under a parent. func GetSubOrderByTokenNetwork(parentTradeId string, token string, network string) (*mdb.Orders, error) { order := new(mdb.Orders) err := dao.Mdb.Model(order). Where("parent_trade_id = ?", parentTradeId). Where("token = ?", token). Where("network = ?", network). Where("status = ?", mdb.StatusWaitPay). Limit(1). Find(order).Error return order, err } // GetSiblingSubOrders returns active sub-orders under the same parent, excluding the given trade_id. func GetSiblingSubOrders(parentTradeId string, excludeTradeId string) ([]mdb.Orders, error) { var orders []mdb.Orders err := dao.Mdb.Model(&mdb.Orders{}). Where("parent_trade_id = ?", parentTradeId). Where("trade_id != ?", excludeTradeId). Where("status = ?", mdb.StatusWaitPay). Find(&orders).Error return orders, err } // MarkParentOrderSuccess marks the parent order as paid and records which sub-order // settled it. Only the status, callback_confirm, and pay_by_sub_id fields are updated; // the parent's own block_transaction_id, actual_amount, and receive_address are // intentionally left unchanged because the parent was not directly paid. func MarkParentOrderSuccess(parentTradeId string, sub *mdb.Orders) (bool, error) { return MarkParentOrderSuccessWithTransaction(dao.Mdb, parentTradeId, sub) } // MarkParentOrderSuccessWithTransaction is the transactional variant of // MarkParentOrderSuccess. func MarkParentOrderSuccessWithTransaction(tx *gorm.DB, parentTradeId string, sub *mdb.Orders) (bool, error) { result := tx.Model(&mdb.Orders{}). Where("trade_id = ?", parentTradeId). Where("status = ?", mdb.StatusWaitPay). Updates(map[string]interface{}{ "status": mdb.StatusPaySuccess, "callback_confirm": mdb.CallBackConfirmNo, "pay_by_sub_id": sub.ID, }) return result.RowsAffected > 0, result.Error } // ExpireSiblingSubOrdersWithTransaction expires all waiting sibling // sub-orders under the same parent in one statement. func ExpireSiblingSubOrdersWithTransaction(tx *gorm.DB, parentTradeId string, excludeTradeId string) error { return tx.Model(&mdb.Orders{}). Where("parent_trade_id = ?", parentTradeId). Where("trade_id != ?", excludeTradeId). Where("status = ?", mdb.StatusWaitPay). Update("status", mdb.StatusExpired).Error } // MarkOrderSelected sets is_selected=true for the given trade_id. func MarkOrderSelected(tradeId string) error { return dao.Mdb.Model(&mdb.Orders{}). Where("trade_id = ?", tradeId). Update("is_selected", true).Error } // ExpireOrderByTradeID marks a waiting order as expired. Used to retire failed // child-order attempts that should not remain selectable/reusable. func ExpireOrderByTradeID(tradeId string) error { return dao.Mdb.Model(&mdb.Orders{}). Where("trade_id = ?", tradeId). Where("status = ?", mdb.StatusWaitPay). Updates(map[string]interface{}{ "status": mdb.StatusExpired, "is_selected": false, }).Error } // RefreshOrderExpiration resets created_at to now so the expiration timer restarts. // Called on the parent order when a sub-order is created or returned. func RefreshOrderExpiration(tradeId string) error { return dao.Mdb.Model(&mdb.Orders{}). Where("trade_id = ?", tradeId). Update("created_at", time.Now()).Error } // ResetCallbackConfirmOk sets callback_confirm back to Ok. // Prevents the callback worker from retrying a sub-order with an empty notify_url. func ResetCallbackConfirmOk(tradeId string) error { return dao.Mdb.Model(&mdb.Orders{}). Where("trade_id = ?", tradeId). Update("callback_confirm", mdb.CallBackConfirmOk).Error } // GetActiveSubOrders returns all active sub-orders under a parent. func GetActiveSubOrders(parentTradeId string) ([]mdb.Orders, error) { var orders []mdb.Orders err := dao.Mdb.Model(&mdb.Orders{}). Where("parent_trade_id = ?", parentTradeId). Where("status = ?", mdb.StatusWaitPay). Find(&orders).Error return orders, err } // GetAllSubOrders returns all sub-orders (any status) under a parent order, // ordered by creation time ascending. func GetAllSubOrders(parentTradeId string) ([]mdb.Orders, error) { var orders []mdb.Orders err := dao.Mdb.Model(&mdb.Orders{}). Where("parent_trade_id = ?", parentTradeId). Order("created_at ASC"). Find(&orders).Error return orders, err } // GetSubOrdersByParentTradeIds batch-fetches all sub-orders (any status) // whose parent_trade_id is in the given set. Ordered by parent_trade_id // then created_at ASC so callers can build a grouped map in one pass. func GetSubOrdersByParentTradeIds(parentTradeIds []string) ([]mdb.Orders, error) { if len(parentTradeIds) == 0 { return nil, nil } var orders []mdb.Orders err := dao.Mdb.Model(&mdb.Orders{}). Where("parent_trade_id IN ?", parentTradeIds). Order("parent_trade_id ASC, created_at ASC"). Find(&orders).Error return orders, err } // ExpireOrderByTradeId marks a single order as expired if still waiting. func ExpireOrderByTradeId(tradeId string) error { return dao.Mdb.Model(&mdb.Orders{}). Where("trade_id = ?", tradeId). Where("status = ?", mdb.StatusWaitPay). Update("status", mdb.StatusExpired).Error } // GetTradeIdByWalletAddressAndAmountAndToken resolves the reserved trade id by network, address, token and amount. func GetTradeIdByWalletAddressAndAmountAndToken(network string, address string, token string, amount float64) (string, error) { network = normalizeLockNetwork(network) address = normalizeLockAddress(network, address) var locks []mdb.TransactionLock err := activeLocksForAddress(dao.RuntimeDB, network, address, token, time.Now()). Order("created_at ASC"). Find(&locks).Error if err != nil { return "", err } for _, lock := range locks { if lockMatchesAmount(lock, amount) { return lock.TradeId, nil } } return "", nil } // LockTransaction reserves a network+address+token+amount pair in sqlite until expiration. func LockTransaction(network, address, token, tradeID string, amount float64, expirationTime time.Duration) error { network = normalizeLockNetwork(network) address = normalizeLockAddress(network, address) precision := GetAmountPrecision() scaledAmount, amountText := normalizeAmount(amount, precision) normalizedToken := normalizeLockToken(token) now := time.Now() lock := &mdb.TransactionLock{ Network: network, Address: address, Token: normalizedToken, AmountScaled: scaledAmount, AmountText: amountText, AmountPrecision: precision, TradeId: tradeID, ExpiresAt: now.Add(expirationTime), } return dao.RuntimeDB.Transaction(func(tx *gorm.DB) error { expiredQuery := tx.Where("network = ?", network). Where("token = ?", normalizedToken). Where("expires_at <= ?", now) expiredQuery = applyLockAddressFilter(expiredQuery, network, address) if err := expiredQuery.Delete(&mdb.TransactionLock{}).Error; err != nil { return err } if err := tx.Where("trade_id = ?", tradeID).Delete(&mdb.TransactionLock{}).Error; err != nil { return err } var existing []mdb.TransactionLock if err := activeLocksForAddress(tx, network, address, normalizedToken, now). Find(&existing).Error; err != nil { return err } candidateAmount := lockAmountDecimal(*lock) for _, item := range existing { if lockAmountDecimal(item).Equal(candidateAmount) { return ErrTransactionLocked } } result := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(lock) if result.Error != nil { return result.Error } if result.RowsAffected == 0 { return ErrTransactionLocked } return nil }) } // UnLockTransaction releases the reservation for network+address+token+amount. func UnLockTransaction(network string, address string, token string, amount float64) error { network = normalizeLockNetwork(network) address = normalizeLockAddress(network, address) var locks []mdb.TransactionLock if err := activeLocksForAddress(dao.RuntimeDB, network, address, token, time.Now()). Find(&locks).Error; err != nil { return err } ids := make([]uint64, 0, len(locks)) for _, lock := range locks { if lockMatchesAmount(lock, amount) { ids = append(ids, lock.ID) } } if len(ids) == 0 { return nil } return dao.RuntimeDB.Where("id IN ?", ids).Delete(&mdb.TransactionLock{}).Error } func UnLockTransactionByTradeId(tradeID string) error { return dao.RuntimeDB.Where("trade_id = ?", tradeID).Delete(&mdb.TransactionLock{}).Error } func CleanupExpiredTransactionLocks() error { return dao.RuntimeDB.Where("expires_at <= ?", time.Now()).Delete(&mdb.TransactionLock{}).Error }