feat: isolate manual verification RPC usage

- add rpc node purpose support for general/manual_verify/both - keep scan, WSS and health checks away from manual_verify nodes - fallback manual hash verification from general RPC to manual_verify RPC - add public cashier hash submission path and errno responses - update tests for RPC selection, failover and manual payment flows
2026-07-07 18:26:16 +00:00 · 2026-05-27 19:36:56 +08:00
parent 13c81ee560
commit c69b0d6c0c
185 changed files with 2525 additions and 483 deletions
@@ -1,14 +1,44 @@
 package data

 import (
+	"fmt"
 	"math/rand"
+	"net/url"
+	"sort"
 	"strings"
+	"sync"
+	"time"

 	"github.com/GMWalletApp/epusdt/model/dao"
 	"github.com/GMWalletApp/epusdt/model/mdb"
 	"github.com/dromara/carbon/v2"
 )

+const (
+	RpcFailoverThreshold = 3
+	RpcFailoverCooldown  = 60 * time.Second
+)
+
+var gRpcFailover = struct {
+	sync.Mutex
+	failures      map[uint64]int
+	cooldownUntil map[uint64]time.Time
+}{
+	failures:      make(map[uint64]int),
+	cooldownUntil: make(map[uint64]time.Time),
+}
+
+func NormalizeRpcNodePurpose(purpose string) string {
+	switch strings.ToLower(strings.TrimSpace(purpose)) {
+	case mdb.RpcNodePurposeManualVerify:
+		return mdb.RpcNodePurposeManualVerify
+	case mdb.RpcNodePurposeBoth:
+		return mdb.RpcNodePurposeBoth
+	default:
+		return mdb.RpcNodePurposeGeneral
+	}
+}
+
 // ListRpcNodes returns rows optionally filtered by network.
 func ListRpcNodes(network string) ([]mdb.RpcNode, error) {
 	var rows []mdb.RpcNode
@@ -20,6 +50,84 @@ func ListRpcNodes(network string) ([]mdb.RpcNode, error) {
 	return rows, err
 }

+// ListRpcNodesForHealth returns nodes that should be probed by the periodic
+// health job. Manual verification nodes are intentionally excluded because
+// they may be paid/high-limit endpoints that should only be used on demand.
+func ListRpcNodesForHealth() ([]mdb.RpcNode, error) {
+	var rows []mdb.RpcNode
+	err := dao.Mdb.Model(&mdb.RpcNode{}).
+		Where("(purpose IN ? OR purpose = '' OR purpose IS NULL)", []string{mdb.RpcNodePurposeGeneral, mdb.RpcNodePurposeBoth}).
+		Order("id ASC").
+		Find(&rows).Error
+	return rows, err
+}
+
+// ListGeneralRpcCandidates returns enabled non-manual nodes for ordinary
+// scanners/listeners. Down nodes are skipped; ok nodes are ordered before
+// unknown/bootstrap nodes.
+func ListGeneralRpcCandidates(network, nodeType string) ([]mdb.RpcNode, error) {
+	var rows []mdb.RpcNode
+	err := dao.Mdb.Model(&mdb.RpcNode{}).
+		Where("network = ?", strings.ToLower(strings.TrimSpace(network))).
+		Where("type = ?", strings.ToLower(strings.TrimSpace(nodeType))).
+		Where("enabled = ?", true).
+		Where("(purpose IN ? OR purpose = '' OR purpose IS NULL)", []string{mdb.RpcNodePurposeGeneral, mdb.RpcNodePurposeBoth}).
+		Where("(status IN ? OR status = '' OR status IS NULL)", []string{mdb.RpcNodeStatusOk, mdb.RpcNodeStatusUnknown}).
+		Find(&rows).Error
+	if err != nil {
+		return nil, err
+	}
+
+	buckets := make([][]mdb.RpcNode, 2)
+	for _, row := range rows {
+		row.Purpose = NormalizeRpcNodePurpose(row.Purpose)
+		if row.Status == mdb.RpcNodeStatusOk {
+			buckets[0] = append(buckets[0], row)
+		} else {
+			buckets[1] = append(buckets[1], row)
+		}
+	}
+
+	out := make([]mdb.RpcNode, 0, len(rows))
+	for _, bucket := range buckets {
+		sortRpcNodes(bucket)
+		out = append(out, bucket...)
+	}
+	return out, nil
+}
+
+func SelectGeneralRpcNode(network, nodeType string, excludeIDs ...uint64) (*mdb.RpcNode, error) {
+	rows, err := ListGeneralRpcCandidates(network, nodeType)
+	if err != nil {
+		return nil, err
+	}
+	if len(rows) == 0 {
+		return nil, nil
+	}
+
+	excluded := make(map[uint64]struct{}, len(excludeIDs))
+	for _, id := range excludeIDs {
+		if id > 0 {
+			excluded[id] = struct{}{}
+		}
+	}
+	for i := range rows {
+		if _, ok := excluded[rows[i].ID]; ok {
+			continue
+		}
+		if IsRpcNodeCoolingDown(rows[i].ID) {
+			continue
+		}
+		return &rows[i], nil
+	}
+	if len(excluded) == 0 {
+		for i := range rows {
+			return &rows[i], nil
+		}
+	}
+	return nil, nil
+}
+
 // GetRpcNodeByID fetches one row.
 func GetRpcNodeByID(id uint64) (*mdb.RpcNode, error) {
 	row := new(mdb.RpcNode)
@@ -29,6 +137,7 @@ func GetRpcNodeByID(id uint64) (*mdb.RpcNode, error) {

 // CreateRpcNode inserts a row.
 func CreateRpcNode(row *mdb.RpcNode) error {
+	row.Purpose = NormalizeRpcNodePurpose(row.Purpose)
 	return dao.Mdb.Create(row).Error
 }

@@ -37,6 +146,9 @@ func UpdateRpcNodeFields(id uint64, fields map[string]interface{}) error {
 	if len(fields) == 0 {
 		return nil
 	}
+	if purpose, ok := fields["purpose"].(string); ok {
+		fields["purpose"] = NormalizeRpcNodePurpose(purpose)
+	}
 	return dao.Mdb.Model(&mdb.RpcNode{}).Where("id = ?", id).Updates(fields).Error
 }

@@ -48,13 +160,15 @@ func DeleteRpcNodeByID(id uint64) error {
 // SelectRpcNode picks a healthy RPC endpoint for a (network, type) pair.
 // Strategy: weighted random among rows where enabled=true AND status=ok.
 // Falls back to enabled rows with status=unknown when no health check has
-// run yet. Explicitly down rows are not selected.
+// run yet. Explicitly down rows are not selected. Manual verification nodes
+// are excluded so paid/manual-only RPCs are never used by scanners/listeners.
 func SelectRpcNode(network, nodeType string) (*mdb.RpcNode, error) {
 	var rows []mdb.RpcNode
 	err := dao.Mdb.Model(&mdb.RpcNode{}).
 		Where("network = ?", strings.ToLower(network)).
 		Where("type = ?", strings.ToLower(nodeType)).
 		Where("enabled = ?", true).
+		Where("(purpose IN ? OR purpose = '' OR purpose IS NULL)", []string{mdb.RpcNodePurposeGeneral, mdb.RpcNodePurposeBoth}).
 		Find(&rows).Error
 	if err != nil {
 		return nil, err
@@ -82,6 +196,144 @@ func SelectRpcNode(network, nodeType string) (*mdb.RpcNode, error) {
 	return pickWeighted(candidates), nil
 }

+// ListManualPaymentRpcCandidates returns enabled RPC nodes for manual payment
+// verification. Ordinary RPCs are tried first to avoid paid endpoint usage;
+// manual_verify nodes are used only as a fallback.
+func ListManualPaymentRpcCandidates(network, nodeType string) ([]mdb.RpcNode, error) {
+	var rows []mdb.RpcNode
+	err := dao.Mdb.Model(&mdb.RpcNode{}).
+		Where("network = ?", strings.ToLower(strings.TrimSpace(network))).
+		Where("type = ?", strings.ToLower(strings.TrimSpace(nodeType))).
+		Where("enabled = ?", true).
+		Where("(status IN ? OR status = '' OR status IS NULL)", []string{mdb.RpcNodeStatusOk, mdb.RpcNodeStatusUnknown}).
+		Find(&rows).Error
+	if err != nil {
+		return nil, err
+	}
+	if len(rows) == 0 {
+		return nil, nil
+	}
+
+	buckets := make([][]mdb.RpcNode, 4)
+	for _, row := range rows {
+		row.Purpose = NormalizeRpcNodePurpose(row.Purpose)
+		switch row.Purpose {
+		case mdb.RpcNodePurposeGeneral, mdb.RpcNodePurposeBoth:
+			if row.Status == mdb.RpcNodeStatusOk {
+				buckets[0] = append(buckets[0], row)
+			} else {
+				buckets[1] = append(buckets[1], row)
+			}
+		case mdb.RpcNodePurposeManualVerify:
+			if row.Status == mdb.RpcNodeStatusOk {
+				buckets[2] = append(buckets[2], row)
+			} else {
+				buckets[3] = append(buckets[3], row)
+			}
+		}
+	}
+
+	out := make([]mdb.RpcNode, 0, len(rows))
+	for _, bucket := range buckets {
+		sortRpcNodes(bucket)
+		out = append(out, bucket...)
+	}
+	return out, nil
+}
+
+func sortRpcNodes(rows []mdb.RpcNode) {
+	sort.SliceStable(rows, func(i, j int) bool {
+		if rows[i].Weight != rows[j].Weight {
+			return rows[i].Weight > rows[j].Weight
+		}
+		return rows[i].ID < rows[j].ID
+	})
+}
+
+func RecordRpcNodeSuccess(nodeID uint64) {
+	if nodeID == 0 {
+		return
+	}
+	gRpcFailover.Lock()
+	defer gRpcFailover.Unlock()
+	delete(gRpcFailover.failures, nodeID)
+	delete(gRpcFailover.cooldownUntil, nodeID)
+}
+
+func RecordRpcNodeFailure(nodeID uint64) (int, bool) {
+	if nodeID == 0 {
+		return 0, false
+	}
+	now := time.Now()
+	gRpcFailover.Lock()
+	defer gRpcFailover.Unlock()
+	if until, ok := gRpcFailover.cooldownUntil[nodeID]; ok {
+		if now.Before(until) {
+			return gRpcFailover.failures[nodeID], true
+		}
+		delete(gRpcFailover.cooldownUntil, nodeID)
+		delete(gRpcFailover.failures, nodeID)
+	}
+
+	gRpcFailover.failures[nodeID]++
+	failures := gRpcFailover.failures[nodeID]
+	if failures >= RpcFailoverThreshold {
+		gRpcFailover.failures[nodeID] = 0
+		gRpcFailover.cooldownUntil[nodeID] = now.Add(RpcFailoverCooldown)
+		return failures, true
+	}
+	return failures, false
+}
+
+func IsRpcNodeCoolingDown(nodeID uint64) bool {
+	if nodeID == 0 {
+		return false
+	}
+	now := time.Now()
+	gRpcFailover.Lock()
+	defer gRpcFailover.Unlock()
+	until, ok := gRpcFailover.cooldownUntil[nodeID]
+	if !ok {
+		return false
+	}
+	if now.Before(until) {
+		return true
+	}
+	delete(gRpcFailover.cooldownUntil, nodeID)
+	delete(gRpcFailover.failures, nodeID)
+	return false
+}
+
+func RpcNodeLogLabel(node mdb.RpcNode) string {
+	return fmt.Sprintf("id=%d network=%s type=%s endpoint=%s", node.ID, node.Network, node.Type, safeRpcEndpoint(node.Url))
+}
+
+func safeRpcEndpoint(raw string) string {
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return ""
+	}
+	parsed, err := url.Parse(raw)
+	if err == nil && parsed.Host != "" {
+		if parsed.Scheme != "" {
+			return parsed.Scheme + "://" + parsed.Host
+		}
+		return parsed.Host
+	}
+	cut := strings.IndexAny(raw, "/?")
+	if cut >= 0 {
+		return raw[:cut]
+	}
+	return raw
+}
+
+func ResetRpcFailoverForTest() {
+	gRpcFailover.Lock()
+	defer gRpcFailover.Unlock()
+	gRpcFailover.failures = make(map[uint64]int)
+	gRpcFailover.cooldownUntil = make(map[uint64]time.Time)
+}
+
 // pickWeighted chooses one row using the Weight column. Weights < 1 are
 // coerced to 1 so an admin-zeroed row still has a chance (use Enabled=false
 // to truly disable).