496 lines
18 KiB
Go
496 lines
18 KiB
Go
package payment
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"math"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp-admin-server/internal/appctx"
|
||
"hyapp-admin-server/internal/config"
|
||
|
||
"go.mongodb.org/mongo-driver/v2/bson"
|
||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||
)
|
||
|
||
// legacyRechargeBillQuery 是 legacy 账单源支持的筛选子集；用户/币商筛选属于 hyapp 语义，legacy 源不适用。
|
||
type legacyRechargeBillQuery struct {
|
||
Keyword string
|
||
RegionID int64
|
||
StartAtMS int64
|
||
EndAtMS int64
|
||
Page int
|
||
PageSize int
|
||
}
|
||
|
||
// RechargeBillSource 是 legacy App（Yumi 等 likei 平台）充值账单的外部只读来源；
|
||
// 命中 app_code 后，财务充值明细和汇总直接走该来源，不再请求 wallet-service。
|
||
type RechargeBillSource interface {
|
||
AppCode() string
|
||
AppName() string
|
||
ListRechargeBills(ctx context.Context, query legacyRechargeBillQuery) ([]rechargeBillDTO, int64, error)
|
||
SummarizeRechargeBills(ctx context.Context, query legacyRechargeBillQuery) (rechargeBillSummaryDTO, error)
|
||
}
|
||
|
||
// legacyRegionCountryResolver 复用财务范围的区域→国家映射（MongoMoneyRegionSource 已实现）。
|
||
type legacyRegionCountryResolver interface {
|
||
CountryCodesForRegion(ctx context.Context, appCode string, regionID int64) ([]string, bool, error)
|
||
}
|
||
|
||
// MongoRechargeBillSource 读取 likei Mongo 的 in_app_purchase_details 集合作为充值账单事实。
|
||
type MongoRechargeBillSource struct {
|
||
collection *mongo.Collection
|
||
config config.FinanceBillSourceConfig
|
||
regionResolvers []legacyRegionCountryResolver
|
||
}
|
||
|
||
// NewMongoRechargeBillSource 创建 legacy 账单源；regionSources 用于把后台区域 ID 解析成国家码过滤条件。
|
||
func NewMongoRechargeBillSource(database *mongo.Database, sourceConfig config.FinanceBillSourceConfig, regionSources ...MoneyRegionSource) *MongoRechargeBillSource {
|
||
if database == nil || !sourceConfig.Enabled {
|
||
return nil
|
||
}
|
||
collectionName := strings.TrimSpace(sourceConfig.MongoCollection)
|
||
if collectionName == "" {
|
||
collectionName = "in_app_purchase_details"
|
||
}
|
||
sourceConfig.AppCode = appctx.Normalize(sourceConfig.AppCode)
|
||
sourceConfig.AppName = strings.TrimSpace(sourceConfig.AppName)
|
||
sourceConfig.SysOrigin = strings.ToUpper(strings.TrimSpace(sourceConfig.SysOrigin))
|
||
source := &MongoRechargeBillSource{collection: database.Collection(collectionName), config: sourceConfig}
|
||
for _, regionSource := range regionSources {
|
||
if resolver, ok := regionSource.(legacyRegionCountryResolver); ok && resolver != nil {
|
||
source.regionResolvers = append(source.regionResolvers, resolver)
|
||
}
|
||
}
|
||
return source
|
||
}
|
||
|
||
func (s *MongoRechargeBillSource) AppCode() string {
|
||
return s.config.AppCode
|
||
}
|
||
|
||
func (s *MongoRechargeBillSource) AppName() string {
|
||
if s.config.AppName != "" {
|
||
return s.config.AppName
|
||
}
|
||
return s.config.AppCode
|
||
}
|
||
|
||
// legacyPurchaseDocument 对应 chatapp3-java 的 InAppPurchaseDetails Mongo 文档（只取展示需要的字段）。
|
||
type legacyPurchaseDocument struct {
|
||
ID any `bson:"_id"`
|
||
OrderID string `bson:"orderId"`
|
||
AcceptUserID int64 `bson:"acceptUserId"`
|
||
CountryCode string `bson:"countryCode"`
|
||
Currency string `bson:"currency"`
|
||
Amount any `bson:"amount"`
|
||
AmountUSD any `bson:"amountUsd"`
|
||
Status string `bson:"status"`
|
||
Factory legacyPurchaseFactory `bson:"factory"`
|
||
Products []legacyPurchaseProduct `bson:"products"`
|
||
CreateTime time.Time `bson:"createTime"`
|
||
PurchaseDateMS time.Time `bson:"purchaseDateMs"`
|
||
}
|
||
|
||
type legacyPurchaseFactory struct {
|
||
Platform string `bson:"platform"`
|
||
FactoryCode string `bson:"factoryCode"`
|
||
}
|
||
|
||
type legacyPurchaseProduct struct {
|
||
Name string `bson:"name"`
|
||
Content string `bson:"content"`
|
||
}
|
||
|
||
func (s *MongoRechargeBillSource) ListRechargeBills(ctx context.Context, query legacyRechargeBillQuery) ([]rechargeBillDTO, int64, error) {
|
||
if s == nil || s.collection == nil {
|
||
return nil, 0, fmt.Errorf("legacy bill source is not configured")
|
||
}
|
||
filter, regionCodes, matchable, err := s.billFilter(ctx, query)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if !matchable {
|
||
return []rechargeBillDTO{}, 0, nil
|
||
}
|
||
page := query.Page
|
||
if page < 1 {
|
||
page = 1
|
||
}
|
||
pageSize := query.PageSize
|
||
if pageSize < 1 {
|
||
pageSize = 10
|
||
}
|
||
if len(regionCodes) == 0 {
|
||
return s.listBillsByFilter(ctx, filter, page, pageSize)
|
||
}
|
||
return s.listBillsByRegionPipeline(ctx, filter, regionCodes, page, pageSize)
|
||
}
|
||
|
||
func (s *MongoRechargeBillSource) listBillsByFilter(ctx context.Context, filter bson.M, page int, pageSize int) ([]rechargeBillDTO, int64, error) {
|
||
total, err := s.collection.CountDocuments(ctx, filter)
|
||
if err != nil {
|
||
return nil, 0, fmt.Errorf("count legacy recharge bills for %s: %w", s.config.AppCode, err)
|
||
}
|
||
cursor, err := s.collection.Find(ctx, filter, options.Find().
|
||
SetSort(bson.D{{Key: "createTime", Value: -1}, {Key: "_id", Value: -1}}).
|
||
SetSkip(int64(page-1)*int64(pageSize)).
|
||
SetLimit(int64(pageSize)))
|
||
if err != nil {
|
||
return nil, 0, fmt.Errorf("query legacy recharge bills for %s: %w", s.config.AppCode, err)
|
||
}
|
||
defer cursor.Close(ctx)
|
||
items, err := s.decodeBillCursor(ctx, cursor, pageSize)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
return items, total, nil
|
||
}
|
||
|
||
// listBillsByRegionPipeline 处理区域筛选：likei 账单的 countryCode 长期为空，区域归属只能按付款用户
|
||
// user_run_profile.countryCode 反查，因此列表和计数都要走 $lookup 聚合。
|
||
func (s *MongoRechargeBillSource) listBillsByRegionPipeline(ctx context.Context, filter bson.M, regionCodes []string, page int, pageSize int) ([]rechargeBillDTO, int64, error) {
|
||
base := append(mongo.Pipeline{bson.D{{Key: "$match", Value: filter}}}, legacyRegionLookupStages(regionCodes)...)
|
||
|
||
countCursor, err := s.collection.Aggregate(ctx, append(clonePipeline(base), bson.D{{Key: "$count", Value: "n"}}))
|
||
if err != nil {
|
||
return nil, 0, fmt.Errorf("count legacy recharge bills for %s: %w", s.config.AppCode, err)
|
||
}
|
||
defer countCursor.Close(ctx)
|
||
var total int64
|
||
if countCursor.Next(ctx) {
|
||
var row struct {
|
||
N int64 `bson:"n"`
|
||
}
|
||
if err := countCursor.Decode(&row); err != nil {
|
||
return nil, 0, fmt.Errorf("decode legacy recharge bill count for %s: %w", s.config.AppCode, err)
|
||
}
|
||
total = row.N
|
||
}
|
||
if err := countCursor.Err(); err != nil {
|
||
return nil, 0, fmt.Errorf("iterate legacy recharge bill count for %s: %w", s.config.AppCode, err)
|
||
}
|
||
if total == 0 {
|
||
return []rechargeBillDTO{}, 0, nil
|
||
}
|
||
|
||
pagePipeline := append(clonePipeline(base),
|
||
bson.D{{Key: "$sort", Value: bson.D{{Key: "createTime", Value: -1}, {Key: "_id", Value: -1}}}},
|
||
bson.D{{Key: "$skip", Value: int64(page-1) * int64(pageSize)}},
|
||
bson.D{{Key: "$limit", Value: int64(pageSize)}},
|
||
)
|
||
cursor, err := s.collection.Aggregate(ctx, pagePipeline)
|
||
if err != nil {
|
||
return nil, 0, fmt.Errorf("query legacy recharge bills for %s: %w", s.config.AppCode, err)
|
||
}
|
||
defer cursor.Close(ctx)
|
||
items, err := s.decodeBillCursor(ctx, cursor, pageSize)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
return items, total, nil
|
||
}
|
||
|
||
func (s *MongoRechargeBillSource) decodeBillCursor(ctx context.Context, cursor *mongo.Cursor, pageSize int) ([]rechargeBillDTO, error) {
|
||
items := make([]rechargeBillDTO, 0, pageSize)
|
||
for cursor.Next(ctx) {
|
||
var document legacyPurchaseDocument
|
||
if err := cursor.Decode(&document); err != nil {
|
||
return nil, fmt.Errorf("decode legacy recharge bill for %s: %w", s.config.AppCode, err)
|
||
}
|
||
items = append(items, legacyRechargeBillDTO(s.config, document))
|
||
}
|
||
if err := cursor.Err(); err != nil {
|
||
return nil, fmt.Errorf("iterate legacy recharge bills for %s: %w", s.config.AppCode, err)
|
||
}
|
||
return items, nil
|
||
}
|
||
|
||
func (s *MongoRechargeBillSource) SummarizeRechargeBills(ctx context.Context, query legacyRechargeBillQuery) (rechargeBillSummaryDTO, error) {
|
||
if s == nil || s.collection == nil {
|
||
return rechargeBillSummaryDTO{}, fmt.Errorf("legacy bill source is not configured")
|
||
}
|
||
filter, regionCodes, matchable, err := s.billFilter(ctx, query)
|
||
if err != nil {
|
||
return rechargeBillSummaryDTO{}, err
|
||
}
|
||
if !matchable {
|
||
return rechargeBillSummaryDTO{}, nil
|
||
}
|
||
// 金币数只统计 GOLD/CANDY 商品内容，口径对齐 chatapp3 的 firstGoldProduct；content 非数值时按 0 计入。
|
||
coinExpr := bson.M{"$reduce": bson.M{
|
||
"input": bson.M{"$filter": bson.M{
|
||
"input": bson.M{"$ifNull": bson.A{"$products", bson.A{}}},
|
||
"cond": bson.M{"$in": bson.A{"$$this.name", bson.A{"GOLD", "CANDY"}}},
|
||
}},
|
||
"initialValue": 0,
|
||
"in": bson.M{"$add": bson.A{
|
||
"$$value",
|
||
bson.M{"$convert": bson.M{"input": "$$this.content", "to": "long", "onError": 0, "onNull": 0}},
|
||
}},
|
||
}}
|
||
pipeline := append(mongo.Pipeline{bson.D{{Key: "$match", Value: filter}}}, legacyRegionLookupStages(regionCodes)...)
|
||
pipeline = append(pipeline, bson.D{{Key: "$group", Value: bson.M{
|
||
"_id": bson.M{"$cond": bson.A{
|
||
bson.M{"$eq": bson.A{"$factory.factoryCode", "GOOGLE"}},
|
||
"google",
|
||
"third_party",
|
||
}},
|
||
"billCount": bson.M{"$sum": 1},
|
||
"usd": bson.M{"$sum": bson.M{"$ifNull": bson.A{"$amountUsd", 0}}},
|
||
"coin": bson.M{"$sum": coinExpr},
|
||
}}})
|
||
cursor, err := s.collection.Aggregate(ctx, pipeline)
|
||
if err != nil {
|
||
return rechargeBillSummaryDTO{}, fmt.Errorf("summarize legacy recharge bills for %s: %w", s.config.AppCode, err)
|
||
}
|
||
defer cursor.Close(ctx)
|
||
|
||
summary := rechargeBillSummaryDTO{}
|
||
for cursor.Next(ctx) {
|
||
var row struct {
|
||
ID string `bson:"_id"`
|
||
BillCount int64 `bson:"billCount"`
|
||
USD any `bson:"usd"`
|
||
Coin int64 `bson:"coin"`
|
||
}
|
||
if err := cursor.Decode(&row); err != nil {
|
||
return rechargeBillSummaryDTO{}, fmt.Errorf("decode legacy recharge summary for %s: %w", s.config.AppCode, err)
|
||
}
|
||
bucket := rechargeBillSummaryBucketDTO{
|
||
BillCount: row.BillCount,
|
||
CoinAmount: row.Coin,
|
||
USDMinorAmount: legacyDecimalToMinor(row.USD),
|
||
}
|
||
if row.ID == "google" {
|
||
summary.GooglePlay = bucket
|
||
} else {
|
||
summary.ThirdParty.BillCount += bucket.BillCount
|
||
summary.ThirdParty.CoinAmount += bucket.CoinAmount
|
||
summary.ThirdParty.USDMinorAmount += bucket.USDMinorAmount
|
||
}
|
||
}
|
||
if err := cursor.Err(); err != nil {
|
||
return rechargeBillSummaryDTO{}, fmt.Errorf("iterate legacy recharge summary for %s: %w", s.config.AppCode, err)
|
||
}
|
||
summary.Total = rechargeBillSummaryBucketDTO{
|
||
BillCount: summary.GooglePlay.BillCount + summary.ThirdParty.BillCount,
|
||
CoinAmount: summary.GooglePlay.CoinAmount + summary.ThirdParty.CoinAmount,
|
||
USDMinorAmount: summary.GooglePlay.USDMinorAmount + summary.ThirdParty.USDMinorAmount,
|
||
}
|
||
return summary, nil
|
||
}
|
||
|
||
// billFilter 构造与 hyapp 账单列表同口径的 Mongo 过滤条件；返回 matchable=false 表示筛选注定无结果。
|
||
// regionCodes 非空时调用方必须叠加 legacyRegionLookupStages：likei 账单不带国家码，区域归属按付款用户资料判定。
|
||
func (s *MongoRechargeBillSource) billFilter(ctx context.Context, query legacyRechargeBillQuery) (bson.M, []string, bool, error) {
|
||
filter := bson.M{
|
||
"sysOrigin": s.config.SysOrigin,
|
||
"status": "SUCCESS",
|
||
// 订阅免费试用不产生真实付款，与 Aslan 大屏充值口径保持一致。
|
||
"trialPeriod": bson.M{"$ne": true},
|
||
}
|
||
timeRange := bson.M{}
|
||
if query.StartAtMS > 0 {
|
||
timeRange["$gte"] = time.UnixMilli(query.StartAtMS)
|
||
}
|
||
if query.EndAtMS > 0 {
|
||
timeRange["$lt"] = time.UnixMilli(query.EndAtMS)
|
||
}
|
||
if len(timeRange) > 0 {
|
||
filter["createTime"] = timeRange
|
||
}
|
||
if keyword := strings.TrimSpace(query.Keyword); keyword != "" {
|
||
// legacy 集合没有 hyapp 交易号；关键词只精确匹配内部订单 ID 和三方订单号，避免大集合正则扫描。
|
||
filter["$or"] = bson.A{
|
||
bson.M{"_id": keyword},
|
||
bson.M{"orderId": keyword},
|
||
}
|
||
}
|
||
if query.RegionID > 0 {
|
||
codes, handled, err := s.regionCountryCodes(ctx, query.RegionID)
|
||
if err != nil {
|
||
return nil, nil, false, err
|
||
}
|
||
if !handled || len(codes) == 0 {
|
||
return nil, nil, false, nil
|
||
}
|
||
return filter, codes, true, nil
|
||
}
|
||
return filter, nil, true, nil
|
||
}
|
||
|
||
const legacyUserProfileCollection = "user_run_profile"
|
||
|
||
// legacyRegionLookupStages 解析账单的区域归属国家：优先账单自身 countryCode（Aslan 的 MifaPay 单会带），
|
||
// 为空时回退付款用户 user_run_profile.countryCode（Yumi/Aslan 的谷歌单都不带国家码）；codes 为空时不追加任何阶段。
|
||
func legacyRegionLookupStages(codes []string) mongo.Pipeline {
|
||
if len(codes) == 0 {
|
||
return mongo.Pipeline{}
|
||
}
|
||
return mongo.Pipeline{
|
||
bson.D{{Key: "$lookup", Value: bson.M{
|
||
"from": legacyUserProfileCollection,
|
||
"localField": "acceptUserId",
|
||
"foreignField": "_id",
|
||
"as": "legacyBillUser",
|
||
}}},
|
||
bson.D{{Key: "$addFields", Value: bson.M{
|
||
"legacyBillCountry": bson.M{"$toUpper": bson.M{"$cond": bson.A{
|
||
bson.M{"$ne": bson.A{bson.M{"$ifNull": bson.A{"$countryCode", ""}}, ""}},
|
||
"$countryCode",
|
||
bson.M{"$ifNull": bson.A{bson.M{"$first": "$legacyBillUser.countryCode"}, ""}},
|
||
}}},
|
||
}}},
|
||
bson.D{{Key: "$match", Value: bson.M{"legacyBillCountry": bson.M{"$in": codes}}}},
|
||
}
|
||
}
|
||
|
||
// clonePipeline 复制聚合管道，避免 append 复用底层数组导致 count/分页两条管道互相踩踏。
|
||
func clonePipeline(pipeline mongo.Pipeline) mongo.Pipeline {
|
||
cloned := make(mongo.Pipeline, len(pipeline))
|
||
copy(cloned, pipeline)
|
||
return cloned
|
||
}
|
||
|
||
func (s *MongoRechargeBillSource) regionCountryCodes(ctx context.Context, regionID int64) ([]string, bool, error) {
|
||
for _, resolver := range s.regionResolvers {
|
||
codes, handled, err := resolver.CountryCodesForRegion(ctx, s.config.AppCode, regionID)
|
||
if err != nil {
|
||
return nil, false, err
|
||
}
|
||
if handled {
|
||
normalized := make([]string, 0, len(codes))
|
||
for _, code := range codes {
|
||
if code = strings.ToUpper(strings.TrimSpace(code)); code != "" {
|
||
normalized = append(normalized, code)
|
||
}
|
||
}
|
||
return normalized, true, nil
|
||
}
|
||
}
|
||
return nil, false, nil
|
||
}
|
||
|
||
// legacyRechargeBillDTO 把 likei 内购明细收敛成财务充值账单 DTO；金额统一按实付币种微单位、美金最小单位表达。
|
||
func legacyRechargeBillDTO(sourceConfig config.FinanceBillSourceConfig, document legacyPurchaseDocument) rechargeBillDTO {
|
||
createdAtMS := int64(0)
|
||
if !document.CreateTime.IsZero() {
|
||
createdAtMS = document.CreateTime.UnixMilli()
|
||
} else if !document.PurchaseDateMS.IsZero() {
|
||
createdAtMS = document.PurchaseDateMS.UnixMilli()
|
||
}
|
||
factoryCode := strings.ToUpper(strings.TrimSpace(document.Factory.FactoryCode))
|
||
dto := rechargeBillDTO{
|
||
AppCode: sourceConfig.AppCode,
|
||
TransactionID: legacyDocumentID(document.ID),
|
||
RechargeType: legacyRechargeType(factoryCode),
|
||
Status: "succeeded",
|
||
ExternalRef: strings.TrimSpace(document.OrderID),
|
||
UserID: document.AcceptUserID,
|
||
CurrencyCode: strings.ToUpper(strings.TrimSpace(document.Currency)),
|
||
CoinAmount: legacyGoldCoinAmount(document.Products),
|
||
USDMinorAmount: legacyDecimalToMinor(document.AmountUSD),
|
||
CreatedAtMS: createdAtMS,
|
||
ProviderCode: legacyProviderCode(factoryCode),
|
||
UserPaidCurrencyCode: strings.ToUpper(strings.TrimSpace(document.Currency)),
|
||
UserPaidAmountMicro: legacyDecimalToMicro(document.Amount),
|
||
// likei 平台没有可反推的三方扣款数据；标记已同步避免前端把 legacy 谷歌账单当成待查询。
|
||
PaidSyncedAtMS: createdAtMS,
|
||
}
|
||
return dto
|
||
}
|
||
|
||
func legacyDocumentID(value any) string {
|
||
if value == nil {
|
||
return ""
|
||
}
|
||
if text, ok := value.(string); ok {
|
||
return text
|
||
}
|
||
if objectID, ok := value.(bson.ObjectID); ok {
|
||
return objectID.Hex()
|
||
}
|
||
return strings.TrimSpace(fmt.Sprint(value))
|
||
}
|
||
|
||
func legacyRechargeType(factoryCode string) string {
|
||
switch factoryCode {
|
||
case "GOOGLE":
|
||
return "google_play_recharge"
|
||
case "APPLE":
|
||
return "apple_recharge"
|
||
case "HUAWEI":
|
||
return "huawei_recharge"
|
||
case "TELEGRAM":
|
||
return "telegram_recharge"
|
||
case "MIFA_PAY":
|
||
// Aslan 的 H5 三方走 MifaPay；复用 hyapp 的 mifapay 展示口径。
|
||
return "mifapay"
|
||
default:
|
||
return "web_recharge"
|
||
}
|
||
}
|
||
|
||
func legacyProviderCode(factoryCode string) string {
|
||
if factoryCode == "" {
|
||
return "legacy"
|
||
}
|
||
return strings.ToLower(factoryCode)
|
||
}
|
||
|
||
// legacyGoldCoinAmount 只统计 GOLD/CANDY 商品内容，口径对齐 chatapp3 的 firstGoldProduct。
|
||
func legacyGoldCoinAmount(products []legacyPurchaseProduct) int64 {
|
||
var total int64
|
||
for _, product := range products {
|
||
if product.Name != "GOLD" && product.Name != "CANDY" {
|
||
continue
|
||
}
|
||
if amount, err := strconv.ParseInt(strings.TrimSpace(product.Content), 10, 64); err == nil {
|
||
total += amount
|
||
}
|
||
}
|
||
return total
|
||
}
|
||
|
||
func legacyDecimalToMicro(value any) int64 {
|
||
return legacyDecimalScaled(value, 1_000_000)
|
||
}
|
||
|
||
func legacyDecimalToMinor(value any) int64 {
|
||
return legacyDecimalScaled(value, 100)
|
||
}
|
||
|
||
// legacyDecimalScaled 把 Mongo 里 Decimal128/数值/字符串形式的十进制金额换算成目标最小单位；非法值按 0 处理。
|
||
func legacyDecimalScaled(value any, scale float64) int64 {
|
||
var text string
|
||
switch typed := value.(type) {
|
||
case nil:
|
||
return 0
|
||
case bson.Decimal128:
|
||
text = typed.String()
|
||
case string:
|
||
text = typed
|
||
case int32:
|
||
return int64(typed) * int64(scale)
|
||
case int64:
|
||
return typed * int64(scale)
|
||
case int:
|
||
return int64(typed) * int64(scale)
|
||
case float64:
|
||
text = strconv.FormatFloat(typed, 'f', -1, 64)
|
||
case float32:
|
||
text = strconv.FormatFloat(float64(typed), 'f', -1, 32)
|
||
default:
|
||
text = fmt.Sprint(value)
|
||
}
|
||
parsed, err := strconv.ParseFloat(strings.TrimSpace(text), 64)
|
||
if err != nil || math.IsNaN(parsed) || math.IsInf(parsed, 0) {
|
||
return 0
|
||
}
|
||
return int64(math.Round(parsed * scale))
|
||
}
|