435 lines
17 KiB
Go
435 lines
17 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"math"
|
||
"strconv"
|
||
"strings"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
"hyapp/services/wallet-service/internal/domain/ledger"
|
||
)
|
||
|
||
// ListRechargeBills 查询 wallet-service 的充值账单投影，后台不能直接读写业务库内部语义。
|
||
// 结果集是两类事实的并集：wallet_recharge_records（用户充值）与 coin_seller_stock_records（币商进货/冲回）。
|
||
func (r *Repository) ListRechargeBills(ctx context.Context, query ledger.ListRechargeBillsQuery) ([]ledger.RechargeBill, int64, error) {
|
||
if r == nil || r.db == nil {
|
||
return nil, 0, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
ctx = contextWithCommandApp(ctx, query.AppCode)
|
||
query = normalizeRechargeBillsQuery(query)
|
||
query.AppCode = appcode.FromContext(ctx)
|
||
|
||
includeRecords, includeStock := rechargeBillBranches(query)
|
||
if !includeRecords && !includeStock {
|
||
return []ledger.RechargeBill{}, 0, nil
|
||
}
|
||
|
||
selects := make([]string, 0, 2)
|
||
args := make([]any, 0, 16)
|
||
var total int64
|
||
if includeRecords {
|
||
where, whereArgs := rechargeBillsWhereSQL(query)
|
||
branchTotal, err := r.countRechargeBillRows(ctx, where, whereArgs...)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
total += branchTotal
|
||
selects = append(selects, rechargeBillRecordsSelectSQL+where)
|
||
args = append(args, whereArgs...)
|
||
}
|
||
if includeStock {
|
||
where, whereArgs := rechargeBillStockWhereSQL(query)
|
||
branchTotal, err := r.countRechargeBillStockRows(ctx, where, whereArgs...)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
total += branchTotal
|
||
selects = append(selects, rechargeBillStockSelectSQL+where)
|
||
args = append(args, whereArgs...)
|
||
}
|
||
|
||
rows, err := r.db.QueryContext(ctx, `
|
||
SELECT * FROM (`+strings.Join(selects, `
|
||
UNION ALL
|
||
`)+`) bills
|
||
ORDER BY created_at_ms DESC, transaction_id DESC
|
||
LIMIT ? OFFSET ?`,
|
||
append(args, query.PageSize, resourceOffset(query.Page, query.PageSize))...,
|
||
)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
items := make([]ledger.RechargeBill, 0, query.PageSize)
|
||
for rows.Next() {
|
||
var item ledger.RechargeBill
|
||
var externalProviderCode, externalCurrencyCode, externalPayloadJSON string
|
||
var paidCurrencyCode string
|
||
var paidAmountMicro, paidTaxMicro, paidNetMicro, paidSyncedAtMS int64
|
||
if err := rows.Scan(
|
||
&item.AppCode,
|
||
&item.TransactionID,
|
||
&item.CommandID,
|
||
&item.RechargeType,
|
||
&item.Status,
|
||
&item.ExternalRef,
|
||
&item.UserID,
|
||
&item.SellerUserID,
|
||
&item.SellerRegionID,
|
||
&item.TargetRegionID,
|
||
&item.PolicyID,
|
||
&item.PolicyVersion,
|
||
&item.CurrencyCode,
|
||
&item.CoinAmount,
|
||
&item.USDMinorAmount,
|
||
&item.ExchangeCoinAmount,
|
||
&item.ExchangeUSDMinorAmount,
|
||
&item.ProviderAmountMinor,
|
||
&item.CreatedAtMS,
|
||
&externalProviderCode,
|
||
&externalCurrencyCode,
|
||
&externalPayloadJSON,
|
||
&paidCurrencyCode,
|
||
&paidAmountMicro,
|
||
&paidTaxMicro,
|
||
&paidNetMicro,
|
||
&paidSyncedAtMS,
|
||
); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
fillRechargeBillPaidFacts(&item, externalProviderCode, externalCurrencyCode, externalPayloadJSON,
|
||
paidCurrencyCode, paidAmountMicro, paidTaxMicro, paidNetMicro, paidSyncedAtMS)
|
||
items = append(items, item)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
return items, total, nil
|
||
}
|
||
|
||
// rechargeBillRecordsSelectSQL 是用户充值分支：external_recharge_orders 补 H5 三方实付，payment_orders 补 Google 实付明细。
|
||
const rechargeBillRecordsSelectSQL = `
|
||
SELECT rr.app_code, rr.transaction_id, wt.command_id, wt.biz_type, wt.status, wt.external_ref,
|
||
rr.user_id, rr.seller_user_id, rr.seller_region_id, rr.target_region_id,
|
||
rr.policy_id, rr.policy_version, rr.currency_code,
|
||
rr.coin_amount, rr.usd_minor_amount, rr.exchange_coin_amount, rr.exchange_usd_minor_amount,
|
||
COALESCE(eo.provider_amount_minor, 0) AS provider_amount_minor, rr.created_at_ms,
|
||
COALESCE(eo.provider_code, '') AS external_provider_code, COALESCE(eo.currency_code, '') AS external_currency_code,
|
||
COALESCE(CAST(eo.provider_payload AS CHAR), '') AS external_payload_json,
|
||
COALESCE(po.paid_currency_code, '') AS paid_currency_code, COALESCE(po.paid_amount_micro, 0) AS paid_amount_micro,
|
||
COALESCE(po.paid_tax_micro, 0) AS paid_tax_micro, COALESCE(po.paid_net_micro, 0) AS paid_net_micro,
|
||
COALESCE(po.paid_synced_at_ms, 0) AS paid_synced_at_ms
|
||
FROM wallet_recharge_records rr
|
||
JOIN wallet_transactions wt ON wt.app_code = rr.app_code AND wt.transaction_id = rr.transaction_id
|
||
LEFT JOIN external_recharge_orders eo ON eo.app_code = rr.app_code AND eo.wallet_transaction_id = rr.transaction_id
|
||
LEFT JOIN payment_orders po ON po.app_code = rr.app_code AND po.wallet_transaction_id = rr.transaction_id
|
||
`
|
||
|
||
// rechargeBillStockSelectSQL 是币商进货分支：列结构与用户充值分支对齐；实付事实来自库存记录本身，
|
||
// 美元口径只认 USDT/USD 付款（micro DIV 10000 = 美分），冲回单的金额与美元都为负数。
|
||
const rechargeBillStockSelectSQL = `
|
||
SELECT sr.app_code, sr.transaction_id, wt.command_id, wt.biz_type, wt.status, wt.external_ref,
|
||
0 AS user_id, sr.seller_user_id,
|
||
COALESCE(CAST(JSON_UNQUOTE(JSON_EXTRACT(sr.metadata_json, '$.seller_region_id')) AS SIGNED), 0) AS seller_region_id,
|
||
0 AS target_region_id,
|
||
0 AS policy_id, '' AS policy_version, sr.paid_currency_code AS currency_code,
|
||
sr.coin_amount, CASE WHEN sr.paid_currency_code IN ('USDT', 'USD') THEN sr.paid_amount_micro DIV 10000 ELSE 0 END AS usd_minor_amount,
|
||
0 AS exchange_coin_amount, 0 AS exchange_usd_minor_amount,
|
||
0 AS provider_amount_minor, sr.created_at_ms,
|
||
'' AS external_provider_code, '' AS external_currency_code, '' AS external_payload_json,
|
||
sr.paid_currency_code AS paid_currency_code, sr.paid_amount_micro AS paid_amount_micro,
|
||
0 AS paid_tax_micro, 0 AS paid_net_micro, sr.created_at_ms AS paid_synced_at_ms
|
||
FROM coin_seller_stock_records sr
|
||
JOIN wallet_transactions wt ON wt.app_code = sr.app_code AND wt.transaction_id = sr.transaction_id
|
||
`
|
||
|
||
// rechargeBillBranches 判定筛选条件命中的账单分支：用户筛选和普通渠道只在充值记录分支，币商进货走库存分支。
|
||
func rechargeBillBranches(query ledger.ListRechargeBillsQuery) (includeRecords bool, includeStock bool) {
|
||
switch query.RechargeType {
|
||
case rechargeTypeFilterCoinSeller:
|
||
includeStock = true
|
||
case "":
|
||
includeRecords = true
|
||
includeStock = true
|
||
default:
|
||
includeRecords = true
|
||
}
|
||
// 库存记录没有充值用户维度；按充值用户筛选时币商分支注定无结果。
|
||
if query.UserID > 0 {
|
||
includeStock = false
|
||
}
|
||
return includeRecords, includeStock
|
||
}
|
||
|
||
const (
|
||
// rechargeTypeFilterCoinSeller/ThirdParty 是后台筛选哨兵值，不对应单一 biz_type：
|
||
// coin_seller 命中币商进货/冲回分支，third_party 命中除 Google 外的所有用户充值渠道。
|
||
rechargeTypeFilterCoinSeller = "coin_seller"
|
||
rechargeTypeFilterThirdParty = "third_party"
|
||
providerCodeCoinSeller = "coin_seller"
|
||
)
|
||
|
||
// fillRechargeBillPaidFacts 把三方订单快照、Google 实付明细和币商进货付款收敛成账单统一的实付字段。
|
||
func fillRechargeBillPaidFacts(item *ledger.RechargeBill, externalProviderCode string, externalCurrencyCode string, externalPayloadJSON string,
|
||
paidCurrencyCode string, paidAmountMicro int64, paidTaxMicro int64, paidNetMicro int64, paidSyncedAtMS int64) {
|
||
if item.RechargeType == bizTypeCoinSellerStockPurchase || item.RechargeType == bizTypeCoinSellerStockDeduction {
|
||
item.ProviderCode = providerCodeCoinSeller
|
||
item.UserPaidCurrencyCode = paidCurrencyCode
|
||
item.UserPaidAmountMicro = paidAmountMicro
|
||
item.PaidSyncedAtMS = paidSyncedAtMS
|
||
return
|
||
}
|
||
if item.RechargeType == bizTypeGooglePlayRecharge {
|
||
item.ProviderCode = "google_play"
|
||
item.UserPaidCurrencyCode = paidCurrencyCode
|
||
item.UserPaidAmountMicro = paidAmountMicro
|
||
item.ProviderTaxMicro = paidTaxMicro
|
||
item.ProviderNetMicro = paidNetMicro
|
||
item.PaidSyncedAtMS = paidSyncedAtMS
|
||
// Google 不单独返回服务费，只能由 实付总额 - 净收入 - 税 推出；任一缺失则保持 0。
|
||
if paidAmountMicro > 0 && paidNetMicro > 0 {
|
||
if fee := paidAmountMicro - paidNetMicro - paidTaxMicro; fee > 0 {
|
||
item.ProviderFeeMicro = fee
|
||
}
|
||
}
|
||
return
|
||
}
|
||
item.ProviderCode = externalProviderCode
|
||
item.UserPaidCurrencyCode = externalCurrencyCode
|
||
if item.ProviderAmountMinor > 0 {
|
||
item.UserPaidAmountMicro = item.ProviderAmountMinor * 10_000
|
||
}
|
||
// V5Pay 的回调和查单快照带 fee/tax/accAmount；MiFaPay 与 USDT 没有该数据，解析不到就保持 0。
|
||
fee, tax, net := externalRechargeSettlementFromPayload(externalPayloadJSON)
|
||
item.ProviderFeeMicro = fee
|
||
item.ProviderTaxMicro = tax
|
||
item.ProviderNetMicro = net
|
||
}
|
||
|
||
// externalRechargeSettlementFromPayload 从 provider_payload 快照解析三方结算字段（实付币种微单位）。
|
||
func externalRechargeSettlementFromPayload(payloadJSON string) (feeMicro int64, taxMicro int64, netMicro int64) {
|
||
payloadJSON = strings.TrimSpace(payloadJSON)
|
||
if payloadJSON == "" || !strings.Contains(payloadJSON, "accAmount") && !strings.Contains(payloadJSON, "\"fee\"") {
|
||
return 0, 0, 0
|
||
}
|
||
var payload struct {
|
||
Fee string `json:"fee"`
|
||
Tax string `json:"tax"`
|
||
AccAmount string `json:"accAmount"`
|
||
}
|
||
if err := json.Unmarshal([]byte(payloadJSON), &payload); err != nil {
|
||
return 0, 0, 0
|
||
}
|
||
return decimalAmountToMicro(payload.Fee), decimalAmountToMicro(payload.Tax), decimalAmountToMicro(payload.AccAmount)
|
||
}
|
||
|
||
// decimalAmountToMicro 把三方返回的十进制金额字符串（如 "10.50"）转成微单位；非法或负数一律按 0 处理。
|
||
func decimalAmountToMicro(value string) int64 {
|
||
value = strings.TrimSpace(value)
|
||
if value == "" {
|
||
return 0
|
||
}
|
||
parsed, err := strconv.ParseFloat(value, 64)
|
||
if err != nil || math.IsNaN(parsed) || math.IsInf(parsed, 0) || parsed <= 0 {
|
||
return 0
|
||
}
|
||
return int64(math.Round(parsed * 1_000_000))
|
||
}
|
||
|
||
// rechargeBillStockWhereSQL 构造币商进货分支的筛选；与列表分支同一套查询参数语义。
|
||
func rechargeBillStockWhereSQL(query ledger.ListRechargeBillsQuery) (string, []any) {
|
||
// 只有 counts_as_seller_recharge 的进货/冲回属于财务充值口径；金币补偿等运营动作不进账单。
|
||
where := `WHERE sr.app_code = ? AND sr.counts_as_seller_recharge = TRUE`
|
||
args := []any{query.AppCode}
|
||
if query.SellerUserID > 0 {
|
||
where += ` AND sr.seller_user_id = ?`
|
||
args = append(args, query.SellerUserID)
|
||
}
|
||
if query.RegionID > 0 {
|
||
where += ` AND COALESCE(CAST(JSON_UNQUOTE(JSON_EXTRACT(sr.metadata_json, '$.seller_region_id')) AS SIGNED), 0) = ?`
|
||
args = append(args, query.RegionID)
|
||
}
|
||
if query.Status != "" {
|
||
where += ` AND wt.status = ?`
|
||
args = append(args, query.Status)
|
||
}
|
||
if query.StartAtMS > 0 {
|
||
where += ` AND sr.created_at_ms >= ?`
|
||
args = append(args, query.StartAtMS)
|
||
}
|
||
if query.EndAtMS > 0 {
|
||
where += ` AND sr.created_at_ms < ?`
|
||
args = append(args, query.EndAtMS)
|
||
}
|
||
if query.Keyword != "" {
|
||
like := "%" + query.Keyword + "%"
|
||
where += ` AND (sr.transaction_id LIKE ? OR wt.command_id LIKE ? OR COALESCE(sr.payment_ref, '') LIKE ? OR sr.evidence_ref LIKE ?)`
|
||
args = append(args, like, like, like, like)
|
||
}
|
||
return where, args
|
||
}
|
||
|
||
func (r *Repository) countRechargeBillStockRows(ctx context.Context, where string, args ...any) (int64, error) {
|
||
var total int64
|
||
err := r.db.QueryRowContext(ctx, `
|
||
SELECT COUNT(*)
|
||
FROM coin_seller_stock_records sr
|
||
JOIN wallet_transactions wt ON wt.app_code = sr.app_code AND wt.transaction_id = sr.transaction_id
|
||
`+where,
|
||
args...,
|
||
).Scan(&total)
|
||
return total, err
|
||
}
|
||
|
||
// SummarizeRechargeBills 按与列表一致的筛选口径聚合充值账单：Google、三方与币商进货三个渠道桶。
|
||
func (r *Repository) SummarizeRechargeBills(ctx context.Context, query ledger.ListRechargeBillsQuery) (ledger.RechargeBillSummary, error) {
|
||
if r == nil || r.db == nil {
|
||
return ledger.RechargeBillSummary{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
ctx = contextWithCommandApp(ctx, query.AppCode)
|
||
query = normalizeRechargeBillsQuery(query)
|
||
query.AppCode = appcode.FromContext(ctx)
|
||
|
||
includeRecords, includeStock := rechargeBillBranches(query)
|
||
var summary ledger.RechargeBillSummary
|
||
if includeRecords {
|
||
if err := r.summarizeRechargeRecordBills(ctx, query, &summary); err != nil {
|
||
return ledger.RechargeBillSummary{}, err
|
||
}
|
||
}
|
||
if includeStock {
|
||
if err := r.summarizeCoinSellerStockBills(ctx, query, &summary); err != nil {
|
||
return ledger.RechargeBillSummary{}, err
|
||
}
|
||
}
|
||
summary.Total = ledger.RechargeBillSummaryBucket{
|
||
BillCount: summary.GooglePlay.BillCount + summary.ThirdParty.BillCount + summary.CoinSeller.BillCount,
|
||
CoinAmount: summary.GooglePlay.CoinAmount + summary.ThirdParty.CoinAmount + summary.CoinSeller.CoinAmount,
|
||
USDMinorAmount: summary.GooglePlay.USDMinorAmount + summary.ThirdParty.USDMinorAmount + summary.CoinSeller.USDMinorAmount,
|
||
}
|
||
return summary, nil
|
||
}
|
||
|
||
func (r *Repository) summarizeRechargeRecordBills(ctx context.Context, query ledger.ListRechargeBillsQuery, summary *ledger.RechargeBillSummary) error {
|
||
where, args := rechargeBillsWhereSQL(query)
|
||
rows, err := r.db.QueryContext(ctx, `
|
||
SELECT wt.biz_type, COUNT(*), COALESCE(SUM(rr.coin_amount), 0), COALESCE(SUM(rr.usd_minor_amount), 0)
|
||
FROM wallet_recharge_records rr
|
||
JOIN wallet_transactions wt ON wt.app_code = rr.app_code AND wt.transaction_id = rr.transaction_id
|
||
`+where+`
|
||
GROUP BY wt.biz_type`,
|
||
args...,
|
||
)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
|
||
for rows.Next() {
|
||
var bizType string
|
||
var bucket ledger.RechargeBillSummaryBucket
|
||
if err := rows.Scan(&bizType, &bucket.BillCount, &bucket.CoinAmount, &bucket.USDMinorAmount); err != nil {
|
||
return err
|
||
}
|
||
if bizType == bizTypeGooglePlayRecharge {
|
||
summary.GooglePlay.BillCount += bucket.BillCount
|
||
summary.GooglePlay.CoinAmount += bucket.CoinAmount
|
||
summary.GooglePlay.USDMinorAmount += bucket.USDMinorAmount
|
||
} else {
|
||
summary.ThirdParty.BillCount += bucket.BillCount
|
||
summary.ThirdParty.CoinAmount += bucket.CoinAmount
|
||
summary.ThirdParty.USDMinorAmount += bucket.USDMinorAmount
|
||
}
|
||
}
|
||
return rows.Err()
|
||
}
|
||
|
||
func (r *Repository) summarizeCoinSellerStockBills(ctx context.Context, query ledger.ListRechargeBillsQuery, summary *ledger.RechargeBillSummary) error {
|
||
where, args := rechargeBillStockWhereSQL(query)
|
||
// 币商口径是进货减冲回的净额：冲回记录的金币与付款额本身就是负数，直接求和。
|
||
err := r.db.QueryRowContext(ctx, `
|
||
SELECT COUNT(*), COALESCE(SUM(sr.coin_amount), 0),
|
||
COALESCE(SUM(CASE WHEN sr.paid_currency_code IN ('USDT', 'USD') THEN sr.paid_amount_micro DIV 10000 ELSE 0 END), 0)
|
||
FROM coin_seller_stock_records sr
|
||
JOIN wallet_transactions wt ON wt.app_code = sr.app_code AND wt.transaction_id = sr.transaction_id
|
||
`+where,
|
||
args...,
|
||
).Scan(&summary.CoinSeller.BillCount, &summary.CoinSeller.CoinAmount, &summary.CoinSeller.USDMinorAmount)
|
||
return err
|
||
}
|
||
|
||
func (r *Repository) countRechargeBillRows(ctx context.Context, where string, args ...any) (int64, error) {
|
||
var total int64
|
||
err := r.db.QueryRowContext(ctx, `
|
||
SELECT COUNT(*)
|
||
FROM wallet_recharge_records rr
|
||
JOIN wallet_transactions wt ON wt.app_code = rr.app_code AND wt.transaction_id = rr.transaction_id
|
||
`+where,
|
||
args...,
|
||
).Scan(&total)
|
||
return total, err
|
||
}
|
||
|
||
func normalizeRechargeBillsQuery(query ledger.ListRechargeBillsQuery) ledger.ListRechargeBillsQuery {
|
||
query.AppCode = appcode.Normalize(query.AppCode)
|
||
query.RechargeType = strings.ToLower(strings.TrimSpace(query.RechargeType))
|
||
query.Status = strings.ToLower(strings.TrimSpace(query.Status))
|
||
query.Keyword = strings.TrimSpace(query.Keyword)
|
||
query.Page, query.PageSize = normalizePage(query.Page, query.PageSize)
|
||
if query.EndAtMS > 0 && query.StartAtMS > query.EndAtMS {
|
||
query.StartAtMS, query.EndAtMS = query.EndAtMS, query.StartAtMS
|
||
}
|
||
return query
|
||
}
|
||
|
||
func rechargeBillsWhereSQL(query ledger.ListRechargeBillsQuery) (string, []any) {
|
||
// 财务系统的 APP 充值详情只展示用户主动向平台充值的账单；币商给用户转普通金币虽然会落充值事实供活动/统计消费，
|
||
// 但它本质是币商出货，不应混入财务充值明细和分页总数。
|
||
where := `WHERE rr.app_code = ? AND wt.biz_type <> ?`
|
||
args := []any{query.AppCode, bizTypeCoinSellerTransfer}
|
||
if query.UserID > 0 {
|
||
where += ` AND rr.user_id = ?`
|
||
args = append(args, query.UserID)
|
||
}
|
||
if query.SellerUserID > 0 {
|
||
where += ` AND rr.seller_user_id = ?`
|
||
args = append(args, query.SellerUserID)
|
||
}
|
||
if query.RegionID > 0 {
|
||
where += ` AND rr.target_region_id = ?`
|
||
args = append(args, query.RegionID)
|
||
}
|
||
switch query.RechargeType {
|
||
case "", rechargeTypeFilterCoinSeller:
|
||
// 空值或币商哨兵不追加渠道条件：分支取舍由 rechargeBillBranches 决定。
|
||
case rechargeTypeFilterThirdParty:
|
||
where += ` AND wt.biz_type <> ?`
|
||
args = append(args, bizTypeGooglePlayRecharge)
|
||
default:
|
||
where += ` AND wt.biz_type = ?`
|
||
args = append(args, query.RechargeType)
|
||
}
|
||
if query.Status != "" {
|
||
where += ` AND wt.status = ?`
|
||
args = append(args, query.Status)
|
||
}
|
||
if query.StartAtMS > 0 {
|
||
where += ` AND rr.created_at_ms >= ?`
|
||
args = append(args, query.StartAtMS)
|
||
}
|
||
if query.EndAtMS > 0 {
|
||
where += ` AND rr.created_at_ms < ?`
|
||
args = append(args, query.EndAtMS)
|
||
}
|
||
if query.Keyword != "" {
|
||
like := "%" + query.Keyword + "%"
|
||
where += ` AND (rr.transaction_id LIKE ? OR wt.command_id LIKE ? OR wt.external_ref LIKE ? OR rr.policy_version LIKE ?)`
|
||
args = append(args, like, like, like, like)
|
||
}
|
||
return where, args
|
||
}
|