276 lines
8.5 KiB
Go
276 lines
8.5 KiB
Go
package financeorder
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"strconv"
|
||
"strings"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"hyapp-admin-server/internal/config"
|
||
)
|
||
|
||
var (
|
||
errLegacyCoinSellerNotFound = errors.New("legacy coin seller not found")
|
||
errLegacyCoinSellerOrderDuplicate = errors.New("legacy coin seller recharge order already exists")
|
||
)
|
||
|
||
type coinSellerTarget struct {
|
||
UserID int64
|
||
DisplayUserID string
|
||
CountryID int64
|
||
RegionID int64
|
||
}
|
||
|
||
type legacyRechargeGrantInput struct {
|
||
OrderID uint
|
||
UserID int64
|
||
ExternalOrderNo string
|
||
ProviderCode string
|
||
Chain string
|
||
USDAmount string
|
||
CoinAmount int64
|
||
OperatorUserID uint
|
||
GrantedAt time.Time
|
||
}
|
||
|
||
type legacyRechargeGrantResult struct {
|
||
TransactionID string
|
||
BalanceAfter int64
|
||
}
|
||
|
||
// LegacyCoinSellerRechargeWriter 把 admin 订单发放到 Aslan/Yumi 独立 legacy 钱包账套；
|
||
// admin 表只保存审计和绑定事实，具体余额/流水仍按旧系统表结构写入。
|
||
type LegacyCoinSellerRechargeWriter interface {
|
||
AppCode() string
|
||
Close() error
|
||
ResolveCoinSeller(ctx context.Context, keyword string) (coinSellerTarget, error)
|
||
GrantCoinSellerRecharge(ctx context.Context, input legacyRechargeGrantInput) (legacyRechargeGrantResult, error)
|
||
}
|
||
|
||
type mysqlLegacyCoinSellerRechargeWriter struct {
|
||
appCode string
|
||
sysOrigin string
|
||
appDB *sql.DB
|
||
walletDB *sql.DB
|
||
requestTimeout time.Duration
|
||
}
|
||
|
||
func NewMySQLLegacyCoinSellerRechargeWriter(source config.LegacyCoinSellerRechargeSourceConfig, appDB *sql.DB, walletDB *sql.DB) LegacyCoinSellerRechargeWriter {
|
||
timeout := source.RequestTimeout
|
||
if timeout <= 0 {
|
||
timeout = 5 * time.Second
|
||
}
|
||
return &mysqlLegacyCoinSellerRechargeWriter{
|
||
appCode: strings.ToLower(strings.TrimSpace(source.AppCode)),
|
||
sysOrigin: strings.ToUpper(strings.TrimSpace(source.SysOrigin)),
|
||
appDB: appDB,
|
||
walletDB: walletDB,
|
||
requestTimeout: timeout,
|
||
}
|
||
}
|
||
|
||
func (w *mysqlLegacyCoinSellerRechargeWriter) AppCode() string {
|
||
if w == nil {
|
||
return ""
|
||
}
|
||
return w.appCode
|
||
}
|
||
|
||
func (w *mysqlLegacyCoinSellerRechargeWriter) Close() error {
|
||
if w == nil {
|
||
return nil
|
||
}
|
||
var first error
|
||
if w.appDB != nil {
|
||
first = w.appDB.Close()
|
||
}
|
||
if w.walletDB != nil {
|
||
if err := w.walletDB.Close(); first == nil {
|
||
first = err
|
||
}
|
||
}
|
||
return first
|
||
}
|
||
|
||
func (w *mysqlLegacyCoinSellerRechargeWriter) ResolveCoinSeller(ctx context.Context, keyword string) (coinSellerTarget, error) {
|
||
if w == nil || w.appDB == nil || w.walletDB == nil {
|
||
return coinSellerTarget{}, errors.New("legacy coin seller recharge writer is not configured")
|
||
}
|
||
user, err := w.resolveLegacyUser(ctx, keyword)
|
||
if err != nil {
|
||
return coinSellerTarget{}, err
|
||
}
|
||
if err := w.requireLegacyCoinSeller(ctx, user.UserID); err != nil {
|
||
return coinSellerTarget{}, err
|
||
}
|
||
return user, nil
|
||
}
|
||
|
||
func (w *mysqlLegacyCoinSellerRechargeWriter) resolveLegacyUser(ctx context.Context, keyword string) (coinSellerTarget, error) {
|
||
keyword = strings.TrimSpace(keyword)
|
||
if keyword == "" {
|
||
return coinSellerTarget{}, errLegacyCoinSellerNotFound
|
||
}
|
||
queryCtx, cancel := context.WithTimeout(ctx, w.requestTimeout)
|
||
defer cancel()
|
||
args := []any{w.sysOrigin, keyword, keyword}
|
||
matchSQL := "(account = ? OR CAST(id AS CHAR) = ?"
|
||
if numeric, err := strconv.ParseInt(keyword, 10, 64); err == nil && numeric > 0 {
|
||
matchSQL += " OR id = ?"
|
||
args = append(args, numeric)
|
||
}
|
||
matchSQL += ")"
|
||
row := w.appDB.QueryRowContext(queryCtx, `
|
||
SELECT id, account, COALESCE(country_id, 0)
|
||
FROM user_base_info
|
||
WHERE origin_sys = ? AND `+matchSQL+` AND COALESCE(is_del, 0) = 0
|
||
ORDER BY id DESC
|
||
LIMIT 1`, args...)
|
||
var target coinSellerTarget
|
||
if err := row.Scan(&target.UserID, &target.DisplayUserID, &target.CountryID); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return coinSellerTarget{}, errLegacyCoinSellerNotFound
|
||
}
|
||
return coinSellerTarget{}, err
|
||
}
|
||
if strings.TrimSpace(target.DisplayUserID) == "" {
|
||
target.DisplayUserID = formatInt64(target.UserID)
|
||
}
|
||
return target, nil
|
||
}
|
||
|
||
func (w *mysqlLegacyCoinSellerRechargeWriter) requireLegacyCoinSeller(ctx context.Context, userID int64) error {
|
||
queryCtx, cancel := context.WithTimeout(ctx, w.requestTimeout)
|
||
defer cancel()
|
||
row := w.walletDB.QueryRowContext(queryCtx, `
|
||
SELECT id
|
||
FROM user_freight_balance
|
||
WHERE sys_origin = ? AND user_id = ? AND COALESCE(is_close, 0) = 0 AND (COALESCE(dealer, 0) = 1 OR COALESCE(super_dealer, 0) = 1)
|
||
LIMIT 1`, w.sysOrigin, userID)
|
||
var id int64
|
||
if err := row.Scan(&id); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return errLegacyCoinSellerNotFound
|
||
}
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (w *mysqlLegacyCoinSellerRechargeWriter) GrantCoinSellerRecharge(ctx context.Context, input legacyRechargeGrantInput) (legacyRechargeGrantResult, error) {
|
||
if w == nil || w.walletDB == nil {
|
||
return legacyRechargeGrantResult{}, errors.New("legacy coin seller recharge writer is not configured")
|
||
}
|
||
queryCtx, cancel := context.WithTimeout(ctx, w.requestTimeout)
|
||
defer cancel()
|
||
tx, err := w.walletDB.BeginTx(queryCtx, &sql.TxOptions{})
|
||
if err != nil {
|
||
return legacyRechargeGrantResult{}, err
|
||
}
|
||
defer func() {
|
||
_ = tx.Rollback()
|
||
}()
|
||
var (
|
||
balanceID int64
|
||
earnPointsText string
|
||
consumptionText string
|
||
)
|
||
if err := tx.QueryRowContext(queryCtx, `
|
||
SELECT id, CAST(COALESCE(earn_points, 0) AS CHAR), CAST(COALESCE(consumption_points, 0) AS CHAR)
|
||
FROM user_freight_balance
|
||
WHERE sys_origin = ? AND user_id = ? AND COALESCE(is_close, 0) = 0 AND (COALESCE(dealer, 0) = 1 OR COALESCE(super_dealer, 0) = 1)
|
||
ORDER BY id DESC
|
||
LIMIT 1
|
||
FOR UPDATE`, w.sysOrigin, input.UserID).Scan(&balanceID, &earnPointsText, &consumptionText); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return legacyRechargeGrantResult{}, errLegacyCoinSellerNotFound
|
||
}
|
||
return legacyRechargeGrantResult{}, err
|
||
}
|
||
var duplicateID int64
|
||
if err := tx.QueryRowContext(queryCtx, `
|
||
SELECT id
|
||
FROM user_freight_balance_running_water
|
||
WHERE sys_origin = ? AND remark = ?
|
||
LIMIT 1`, w.sysOrigin, input.ExternalOrderNo).Scan(&duplicateID); err != nil && !errors.Is(err, sql.ErrNoRows) {
|
||
return legacyRechargeGrantResult{}, err
|
||
} else if duplicateID > 0 {
|
||
return legacyRechargeGrantResult{}, errLegacyCoinSellerOrderDuplicate
|
||
}
|
||
if _, err := tx.ExecContext(queryCtx, `
|
||
UPDATE user_freight_balance
|
||
SET earn_points = COALESCE(earn_points, 0) + ?, update_user = ?, update_time = ?
|
||
WHERE id = ?`, input.CoinAmount, input.UserID, input.GrantedAt, balanceID); err != nil {
|
||
return legacyRechargeGrantResult{}, err
|
||
}
|
||
balanceAfter, err := decimalBalanceAfter(earnPointsText, consumptionText, input.CoinAmount)
|
||
if err != nil {
|
||
return legacyRechargeGrantResult{}, err
|
||
}
|
||
waterID := legacySnowflakeLikeID(input.GrantedAt)
|
||
rechargeType := legacyRechargeType(input.ProviderCode, input.Chain)
|
||
if _, err := tx.ExecContext(queryCtx, `
|
||
INSERT INTO user_freight_balance_running_water
|
||
(id, sys_origin, user_id, accept_user_id, seller_id, type, quantity, usd_quantity, balance, amount, recharge_type, origin, remark, create_user, update_user, create_time, update_time)
|
||
VALUES
|
||
(?, ?, ?, ?, ?, 0, ?, 0, ?, ?, ?, 'PURCHASE', ?, ?, ?, ?, ?)`,
|
||
waterID,
|
||
w.sysOrigin,
|
||
input.UserID,
|
||
input.UserID,
|
||
nil,
|
||
input.CoinAmount,
|
||
balanceAfter,
|
||
input.USDAmount,
|
||
rechargeType,
|
||
input.ExternalOrderNo,
|
||
input.UserID,
|
||
input.UserID,
|
||
input.GrantedAt,
|
||
input.GrantedAt,
|
||
); err != nil {
|
||
return legacyRechargeGrantResult{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return legacyRechargeGrantResult{}, err
|
||
}
|
||
return legacyRechargeGrantResult{
|
||
TransactionID: fmt.Sprintf("legacy-freight-purchase-%d", waterID),
|
||
BalanceAfter: balanceAfter,
|
||
}, nil
|
||
}
|
||
|
||
func decimalBalanceAfter(earnPointsText string, consumptionText string, delta int64) (int64, error) {
|
||
earn, err := strconv.ParseFloat(strings.TrimSpace(earnPointsText), 64)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
consumption, err := strconv.ParseFloat(strings.TrimSpace(consumptionText), 64)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return int64(earn) + delta - int64(consumption), nil
|
||
}
|
||
|
||
func legacyRechargeType(providerCode string, chain string) string {
|
||
providerCode = strings.ToUpper(strings.TrimSpace(providerCode))
|
||
if providerCode == "USDT" {
|
||
if chain = strings.ToUpper(strings.TrimSpace(chain)); chain != "" {
|
||
return "USDT-" + chain + "-进货"
|
||
}
|
||
return "USDT-进货"
|
||
}
|
||
return providerCode + "-进货"
|
||
}
|
||
|
||
func legacySnowflakeLikeID(now time.Time) int64 {
|
||
// 旧服务用 MyBatis ASSIGN_ID 写 bigint 主键；后台没有接入旧雪花 worker，只需保证单进程内足够唯一且单调。
|
||
return now.UTC().UnixMilli()*1000 + int64(legacyWaterIDCounter.Add(1)%1000)
|
||
}
|
||
|
||
var legacyWaterIDCounter atomic.Uint32
|