package payment import ( "context" "errors" "fmt" "math" "sort" "strconv" "strings" "time" "hyapp-admin-server/internal/appctx" "hyapp-admin-server/internal/config" "hyapp-admin-server/internal/integration/googleorders" "hyapp-admin-server/internal/model" "hyapp-admin-server/internal/repository" "go.mongodb.org/mongo-driver/v2/bson" "go.mongodb.org/mongo-driver/v2/mongo" ) // legacyRechargeBillQuery 是 legacy 账单源支持的筛选子集;用户/币商筛选属于 hyapp 语义,legacy 源不适用。 type legacyRechargeBillQuery struct { Keyword string // RechargeType 支持 google_play_recharge / third_party 两个口径;legacy 平台没有币商分支。 RechargeType string // PaidState=unsynced 只看谷歌实付未同步的账单。 PaidState 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) // SupportsGooglePaidSync 为 true 时,RefreshGooglePaidDetails 用该 App 的 Play Console 服务账号同步实付明细。 SupportsGooglePaidSync() bool RefreshGooglePaidDetails(ctx context.Context, transactionIDs []string) ([]googleRechargePaidDTO, error) // Overview 返回财务概览聚合:按日趋势、区域分布与谷歌实付覆盖度。 Overview(ctx context.Context, query legacyRechargeBillQuery, tzOffsetMinutes int32) (rechargeBillOverviewDTO, error) } // legacyRegionCountryResolver 复用财务范围的区域→国家映射(MongoMoneyRegionSource 已实现)。 type legacyRegionCountryResolver interface { CountryCodesForRegion(ctx context.Context, appCode string, regionID int64) ([]string, bool, error) } // LegacyGooglePaidStore 缓存 legacy 谷歌账单的 Orders API 实付明细(repository.Store 实现)。 type LegacyGooglePaidStore interface { UpsertLegacyGooglePaidDetail(detail model.LegacyGooglePaidDetail) error LegacyGooglePaidDetails(appCode string, transactionIDs []string) (map[string]model.LegacyGooglePaidDetail, error) LegacyGooglePaidStatsByBillTime(appCode string, startAtMS int64, endAtMS int64) (repository.LegacyGooglePaidStats, error) LegacyGooglePaidTransactionIDs(appCode string) ([]string, error) } // legacyGoogleOrdersClient 按 legacy App 各自的 Play Console 服务账号查订单实付。 type legacyGoogleOrdersClient interface { GetOrder(ctx context.Context, orderID string) (googleorders.Order, error) } // errLegacyGooglePaidUnsupported 表示该账单源未配置谷歌包名/服务账号或实付缓存存储。 var errLegacyGooglePaidUnsupported = errors.New("legacy google paid sync is not configured") // MongoRechargeBillSource 读取 likei Mongo 的 in_app_purchase_details 集合作为充值账单事实。 type MongoRechargeBillSource struct { collection *mongo.Collection database *mongo.Database config config.FinanceBillSourceConfig regionResolvers []legacyRegionCountryResolver regionSources []MoneyRegionSource paidStore LegacyGooglePaidStore googleOrders legacyGoogleOrdersClient } // 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), database: database, config: sourceConfig} for _, regionSource := range regionSources { if regionSource == nil { continue } source.regionSources = append(source.regionSources, regionSource) if resolver, ok := regionSource.(legacyRegionCountryResolver); ok { source.regionResolvers = append(source.regionResolvers, resolver) } } return source } // SetGooglePaidSync 注入实付缓存存储与该 App 的 Google Orders 客户端;未注入时“查询谷歌实付”对该源不可用。 func (s *MongoRechargeBillSource) SetGooglePaidSync(paidStore LegacyGooglePaidStore, googleOrders legacyGoogleOrdersClient) { if s == nil { return } s.paidStore = paidStore s.googleOrders = googleOrders } // SupportsGooglePaidSync 供 handler 判定该源能否响应谷歌实付刷新。 func (s *MongoRechargeBillSource) SupportsGooglePaidSync() bool { return s != nil && s.paidStore != nil && s.googleOrders != nil } 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"` // LegacyBillCountry 由聚合阶段(legacyCountryResolveStages)解析:账单国家码优先,回退付款用户资料。 LegacyBillCountry string `bson:"legacyBillCountry"` } 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) } // 国家解析放在分页之后:每页最多 pageSize 次 $lookup,用于把账单落到区域列展示。 pipeline := append(mongo.Pipeline{ bson.D{{Key: "$match", Value: filter}}, 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)}}, }, legacyCountryResolveStages()...) cursor, err := s.collection.Aggregate(ctx, pipeline) 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) { countryToRegion := s.legacyCountryRegionIndex(ctx) 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) } item := legacyRechargeBillDTO(s.config, document) // likei 账单没有区域 ID;用解析出的国家码映射区域目录,供列表区域列与导出展示。 if country := strings.ToUpper(strings.TrimSpace(document.LegacyBillCountry)); country != "" { if region, ok := countryToRegion[country]; ok { item.TargetRegionID = region.RegionID } } items = append(items, item) } if err := cursor.Err(); err != nil { return nil, fmt.Errorf("iterate legacy recharge bills for %s: %w", s.config.AppCode, err) } if err := s.fillGooglePaidDetails(items); err != nil { return nil, err } return items, nil } // fillGooglePaidDetails 把已同步的 Orders API 实付明细回填到当页谷歌账单:实付币种/总额/税/净收入, // 手续费与 Lalu 同口径反推(实付 − 净收入 − 税)。未同步的谷歌账单保持 paidSyncedAtMs=0,前端提示可查询。 func (s *MongoRechargeBillSource) fillGooglePaidDetails(items []rechargeBillDTO) error { if s.paidStore == nil { return nil } googleTxIDs := make([]string, 0, len(items)) for _, item := range items { if item.RechargeType == "google_play_recharge" { googleTxIDs = append(googleTxIDs, item.TransactionID) } } if len(googleTxIDs) == 0 { return nil } details, err := s.paidStore.LegacyGooglePaidDetails(s.config.AppCode, googleTxIDs) if err != nil { return fmt.Errorf("load legacy google paid details for %s: %w", s.config.AppCode, err) } for index := range items { detail, ok := details[items[index].TransactionID] if !ok || items[index].RechargeType != "google_play_recharge" { continue } items[index].UserPaidCurrencyCode = detail.PaidCurrencyCode items[index].UserPaidAmountMicro = detail.PaidAmountMicro items[index].ProviderTaxMicro = detail.PaidTaxMicro items[index].ProviderNetMicro = detail.PaidNetMicro if detail.PaidAmountMicro > 0 && detail.PaidNetMicro > 0 { if fee := detail.PaidAmountMicro - detail.PaidNetMicro - detail.PaidTaxMicro; fee > 0 { items[index].ProviderFeeMicro = fee } } items[index].PaidSyncedAtMS = detail.SyncedAtMS } return nil } const maxLegacyGooglePaidRefreshBatch = 100 // RefreshGooglePaidDetails 用该 App 的 Play Console 服务账号逐单拉取实付明细并缓存;单笔失败不阻断整批。 func (s *MongoRechargeBillSource) RefreshGooglePaidDetails(ctx context.Context, transactionIDs []string) ([]googleRechargePaidDTO, error) { if !s.SupportsGooglePaidSync() { return nil, errLegacyGooglePaidUnsupported } cleaned := make([]string, 0, len(transactionIDs)) for _, transactionID := range transactionIDs { if transactionID = strings.TrimSpace(transactionID); transactionID != "" { cleaned = append(cleaned, transactionID) } } if len(cleaned) == 0 { return []googleRechargePaidDTO{}, nil } if len(cleaned) > maxLegacyGooglePaidRefreshBatch { return nil, fmt.Errorf("too many transaction_ids") } idValues := make(bson.A, 0, len(cleaned)) for _, transactionID := range cleaned { idValues = append(idValues, transactionID) } cursor, err := s.collection.Find(ctx, bson.M{ "_id": bson.M{"$in": idValues}, "sysOrigin": s.config.SysOrigin, }) if err != nil { return nil, fmt.Errorf("load legacy bills for google paid refresh: %w", err) } defer cursor.Close(ctx) bills := map[string]legacyPurchaseDocument{} for cursor.Next(ctx) { var document legacyPurchaseDocument if err := cursor.Decode(&document); err != nil { return nil, fmt.Errorf("decode legacy bill for google paid refresh: %w", err) } bills[legacyDocumentID(document.ID)] = document } if err := cursor.Err(); err != nil { return nil, fmt.Errorf("iterate legacy bills for google paid refresh: %w", err) } results := make([]googleRechargePaidDTO, 0, len(cleaned)) for _, transactionID := range cleaned { results = append(results, s.refreshGooglePaidDetail(ctx, transactionID, bills)) } return results, nil } func (s *MongoRechargeBillSource) refreshGooglePaidDetail(ctx context.Context, transactionID string, bills map[string]legacyPurchaseDocument) googleRechargePaidDTO { result := googleRechargePaidDTO{TransactionID: transactionID} document, ok := bills[transactionID] if !ok { result.Error = "账单不存在" return result } if strings.ToUpper(strings.TrimSpace(document.Factory.FactoryCode)) != "GOOGLE" { result.Error = "不是谷歌账单" return result } orderID := strings.TrimSpace(document.OrderID) if orderID == "" { result.Error = "缺少谷歌订单号" return result } order, err := s.googleOrders.GetOrder(ctx, orderID) if err != nil { result.Error = err.Error() return result } syncedAt := time.Now().UnixMilli() billCreatedAtMS := int64(0) if !document.CreateTime.IsZero() { billCreatedAtMS = document.CreateTime.UnixMilli() } else if !document.PurchaseDateMS.IsZero() { billCreatedAtMS = document.PurchaseDateMS.UnixMilli() } if err := s.paidStore.UpsertLegacyGooglePaidDetail(model.LegacyGooglePaidDetail{ AppCode: s.config.AppCode, TransactionID: transactionID, ProviderOrderID: orderID, PaidCurrencyCode: strings.ToUpper(strings.TrimSpace(order.CurrencyCode)), PaidAmountMicro: order.TotalAmountMicro, PaidTaxMicro: order.TaxAmountMicro, PaidNetMicro: order.NetAmountMicro, BillUSDMinor: legacyDecimalToMinor(document.AmountUSD), BillCreatedAtMS: billCreatedAtMS, SyncedAtMS: syncedAt, }); err != nil { result.Error = err.Error() return result } result.CurrencyCode = strings.ToUpper(strings.TrimSpace(order.CurrencyCode)) result.PaidAmountMicro = order.TotalAmountMicro result.TaxMicro = order.TaxAmountMicro result.NetMicro = order.NetAmountMicro result.SyncedAtMS = syncedAt return result } 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}, } } switch strings.ToLower(strings.TrimSpace(query.RechargeType)) { case "google_play_recharge": filter["factory.factoryCode"] = "GOOGLE" case "third_party": filter["factory.factoryCode"] = bson.M{"$ne": "GOOGLE"} case "coin_seller": // legacy 平台没有币商进货口径。 return nil, nil, false, nil } if strings.ToLower(strings.TrimSpace(query.PaidState)) == "unsynced" { filter["factory.factoryCode"] = "GOOGLE" if s.paidStore != nil { syncedIDs, err := s.paidStore.LegacyGooglePaidTransactionIDs(s.config.AppCode) if err != nil { return nil, nil, false, fmt.Errorf("load synced legacy google tx ids for %s: %w", s.config.AppCode, err) } if len(syncedIDs) > 0 { idValues := make(bson.A, 0, len(syncedIDs)) for _, transactionID := range syncedIDs { idValues = append(idValues, transactionID) } filter["_id"] = bson.M{"$nin": idValues} } } } 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" // legacyCountryResolveStages 解析账单的区域归属国家:优先账单自身 countryCode(Aslan 的 MifaPay 单会带), // 为空时回退付款用户 user_run_profile.countryCode(Yumi/Aslan 的谷歌单都不带国家码)。 func legacyCountryResolveStages() 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"}, ""}}, }}}, }}}, } } // legacyRegionLookupStages 在国家解析的基础上按区域国家清单过滤;codes 为空时不追加任何阶段。 func legacyRegionLookupStages(codes []string) mongo.Pipeline { if len(codes) == 0 { return mongo.Pipeline{} } return append(legacyCountryResolveStages(), 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), // 非谷歌渠道(MifaPay 等)没有可反推的三方扣款数据,标记已同步; // 谷歌账单保持未同步(0),由 Orders API 实付缓存回填后才算同步完成。 PaidSyncedAtMS: createdAtMS, } if factoryCode == "GOOGLE" { dto.PaidSyncedAtMS = 0 } 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 } // Overview 返回 legacy 账单的按日趋势、区域分布与谷歌实付覆盖度;tzOffsetMinutes 决定日边界。 func (s *MongoRechargeBillSource) Overview(ctx context.Context, query legacyRechargeBillQuery, tzOffsetMinutes int32) (rechargeBillOverviewDTO, error) { overview := rechargeBillOverviewDTO{Daily: []rechargeBillDailyBucketDTO{}, Regions: []rechargeBillRegionBucketDTO{}} if s == nil || s.collection == nil { return overview, fmt.Errorf("legacy bill source is not configured") } // 用户提现(银行卡余额兑换金币/转货运代理)与充值渠道筛选无关,先于渠道可匹配性判断统计。 if err := s.overviewWithdrawal(ctx, query, &overview); err != nil { return overview, err } filter, regionCodes, matchable, err := s.billFilter(ctx, query) if err != nil { return overview, err } if !matchable { return overview, nil } base := append(mongo.Pipeline{bson.D{{Key: "$match", Value: filter}}}, legacyRegionLookupStages(regionCodes)...) if err := s.overviewDaily(ctx, base, tzOffsetMinutes, &overview); err != nil { return overview, err } if err := s.overviewRegions(ctx, base, &overview); err != nil { return overview, err } if err := s.overviewGooglePaid(ctx, filter, query, &overview); err != nil { return overview, err } return overview, nil } func (s *MongoRechargeBillSource) overviewDaily(ctx context.Context, base mongo.Pipeline, tzOffsetMinutes int32, overview *rechargeBillOverviewDTO) error { 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}}, }}, }} isGoogle := bson.M{"$eq": bson.A{"$factory.factoryCode", "GOOGLE"}} pipeline := append(clonePipeline(base), bson.D{{Key: "$group", Value: bson.M{ "_id": bson.M{"$dateToString": bson.M{"format": "%Y-%m-%d", "date": "$createTime", "timezone": legacyTimezoneOffset(tzOffsetMinutes)}}, "googleUsd": bson.M{"$sum": bson.M{"$cond": bson.A{isGoogle, bson.M{"$ifNull": bson.A{"$amountUsd", 0}}, 0}}}, "thirdUsd": bson.M{"$sum": bson.M{"$cond": bson.A{isGoogle, 0, bson.M{"$ifNull": bson.A{"$amountUsd", 0}}}}}, "googleCoin": bson.M{"$sum": bson.M{"$cond": bson.A{isGoogle, coinExpr, 0}}}, "thirdCoin": bson.M{"$sum": bson.M{"$cond": bson.A{isGoogle, 0, coinExpr}}}, }}}, bson.D{{Key: "$sort", Value: bson.M{"_id": 1}}}, ) cursor, err := s.collection.Aggregate(ctx, pipeline) if err != nil { return fmt.Errorf("aggregate legacy daily overview for %s: %w", s.config.AppCode, err) } defer cursor.Close(ctx) for cursor.Next(ctx) { var row struct { Date string `bson:"_id"` GoogleUSD any `bson:"googleUsd"` ThirdUSD any `bson:"thirdUsd"` GoogleCoin int64 `bson:"googleCoin"` ThirdCoin int64 `bson:"thirdCoin"` } if err := cursor.Decode(&row); err != nil { return fmt.Errorf("decode legacy daily overview for %s: %w", s.config.AppCode, err) } overview.Daily = append(overview.Daily, rechargeBillDailyBucketDTO{ Date: row.Date, GoogleUsdMinor: legacyDecimalToMinor(row.GoogleUSD), ThirdPartyUsdMinor: legacyDecimalToMinor(row.ThirdUSD), GoogleCoinAmount: row.GoogleCoin, ThirdPartyCoinAmount: row.ThirdCoin, }) } return cursor.Err() } func (s *MongoRechargeBillSource) overviewRegions(ctx context.Context, base mongo.Pipeline, overview *rechargeBillOverviewDTO) error { pipeline := clonePipeline(base) // 区域分布需要国家解析;若区域筛选未触发 lookup,这里补上(重复 addFields 无害但避免重复 lookup)。 if len(pipeline) == 1 { pipeline = append(pipeline, legacyCountryResolveStages()...) } pipeline = append(pipeline, bson.D{{Key: "$group", Value: bson.M{ "_id": "$legacyBillCountry", "billCount": bson.M{"$sum": 1}, "usd": bson.M{"$sum": bson.M{"$ifNull": bson.A{"$amountUsd", 0}}}, }}}, ) cursor, err := s.collection.Aggregate(ctx, pipeline) if err != nil { return fmt.Errorf("aggregate legacy region overview for %s: %w", s.config.AppCode, err) } defer cursor.Close(ctx) countryToRegion := s.legacyCountryRegionIndex(ctx) buckets := map[int64]*rechargeBillRegionBucketDTO{} for cursor.Next(ctx) { var row struct { Country string `bson:"_id"` BillCount int64 `bson:"billCount"` USD any `bson:"usd"` } if err := cursor.Decode(&row); err != nil { return fmt.Errorf("decode legacy region overview for %s: %w", s.config.AppCode, err) } region, ok := countryToRegion[strings.ToUpper(strings.TrimSpace(row.Country))] if !ok { continue } bucket, exists := buckets[region.RegionID] if !exists { bucket = &rechargeBillRegionBucketDTO{RegionID: region.RegionID, Name: region.Name} buckets[region.RegionID] = bucket } bucket.BillCount += row.BillCount bucket.UsdMinorAmount += legacyDecimalToMinor(row.USD) } if err := cursor.Err(); err != nil { return err } for _, bucket := range buckets { overview.Regions = append(overview.Regions, *bucket) } sort.Slice(overview.Regions, func(i, j int) bool { return overview.Regions[i].UsdMinorAmount > overview.Regions[j].UsdMinorAmount }) return nil } type legacyRegionRef struct { RegionID int64 Name string } // legacyCountryRegionIndex 从区域目录构建国家→区域映射;目录读取失败时区域分布降级为空,不阻断概览其他数据。 func (s *MongoRechargeBillSource) legacyCountryRegionIndex(ctx context.Context) map[string]legacyRegionRef { index := map[string]legacyRegionRef{} for _, regionSource := range s.regionSources { _, regions, _, err := regionSource.ListMoneyMasterData(ctx, repository.MoneyAccess{All: true}) if err != nil { continue } for _, region := range regions { if appctx.Normalize(region.AppCode) != s.config.AppCode { continue } for _, country := range region.Countries { country = strings.ToUpper(strings.TrimSpace(country)) if country == "" { continue } if _, exists := index[country]; !exists { index[country] = legacyRegionRef{RegionID: region.RegionID, Name: region.Name} } } } } return index } // legacyGoldWithdrawCollection 是 likei 的“银行卡余额兑换金币”申请集合; // 该平台的用户工资转币商即此流程(币商在 likei 叫货运代理 FREIGHT),amount 为提交的美元余额。 const legacyGoldWithdrawCollection = "user_bank_withdraw_gold_apply" func (s *MongoRechargeBillSource) overviewWithdrawal(ctx context.Context, query legacyRechargeBillQuery, overview *rechargeBillOverviewDTO) error { if s.database == nil { return nil } match := bson.M{"sysOrigin": s.config.SysOrigin} 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 { match["createTime"] = timeRange } pipeline := mongo.Pipeline{bson.D{{Key: "$match", Value: match}}} if query.RegionID > 0 { codes, handled, err := s.regionCountryCodes(ctx, query.RegionID) if err != nil { return err } if !handled || len(codes) == 0 { return nil } // 兑换申请没有国家码,按提交用户的资料国家落区域。 pipeline = append(pipeline, bson.D{{Key: "$lookup", Value: bson.M{ "from": legacyUserProfileCollection, "localField": "submitUserId", "foreignField": "_id", "as": "legacyBillUser", }}}, bson.D{{Key: "$addFields", Value: bson.M{ "legacyBillCountry": bson.M{"$toUpper": bson.M{"$ifNull": bson.A{ bson.M{"$first": "$legacyBillUser.countryCode"}, "", }}}, }}}, bson.D{{Key: "$match", Value: bson.M{"legacyBillCountry": bson.M{"$in": codes}}}}, ) } pipeline = append(pipeline, bson.D{{Key: "$group", Value: bson.M{ "_id": nil, "n": bson.M{"$sum": 1}, "amount": bson.M{"$sum": bson.M{"$ifNull": bson.A{"$amount", 0}}}, }}}) cursor, err := s.database.Collection(legacyGoldWithdrawCollection).Aggregate(ctx, pipeline) if err != nil { return fmt.Errorf("aggregate legacy gold withdraw for %s: %w", s.config.AppCode, err) } defer cursor.Close(ctx) if cursor.Next(ctx) { var row struct { N int64 `bson:"n"` Amount any `bson:"amount"` } if err := cursor.Decode(&row); err != nil { return fmt.Errorf("decode legacy gold withdraw for %s: %w", s.config.AppCode, err) } overview.Withdrawal.TransferCount = row.N overview.Withdrawal.TransferUsdMinor = legacyDecimalToMinor(row.Amount) } return cursor.Err() } func (s *MongoRechargeBillSource) overviewGooglePaid(ctx context.Context, filter bson.M, query legacyRechargeBillQuery, overview *rechargeBillOverviewDTO) error { rechargeType := strings.ToLower(strings.TrimSpace(query.RechargeType)) if rechargeType != "" && rechargeType != "google_play_recharge" { return nil } googleFilter := bson.M{} for key, value := range filter { googleFilter[key] = value } googleFilter["factory.factoryCode"] = "GOOGLE" googleCount, err := s.collection.CountDocuments(ctx, googleFilter) if err != nil { return fmt.Errorf("count legacy google bills for %s: %w", s.config.AppCode, err) } overview.GooglePaid.GoogleBillCount = googleCount if s.paidStore == nil { overview.GooglePaid.UnsyncedCount = googleCount return nil } stats, err := s.paidStore.LegacyGooglePaidStatsByBillTime(s.config.AppCode, query.StartAtMS, query.EndAtMS) if err != nil { return fmt.Errorf("load legacy google paid stats for %s: %w", s.config.AppCode, err) } overview.GooglePaid.SyncedCount = stats.SyncedCount if unsynced := googleCount - stats.SyncedCount; unsynced > 0 { overview.GooglePaid.UnsyncedCount = unsynced } overview.GooglePaid.CoveredUsdMinor = stats.CoveredUSDMinor overview.GooglePaid.EstFeeUsdMinor = stats.EstFeeUSDMinor overview.GooglePaid.EstTaxUsdMinor = stats.EstTaxUSDMinor overview.GooglePaid.EstNetUsdMinor = stats.EstNetUSDMinor return nil } // legacyTimezoneOffset 把分钟偏移转成 Mongo $dateToString 的时区串,如 480 → "+08:00"。 func legacyTimezoneOffset(tzOffsetMinutes int32) string { sign := "+" minutes := tzOffsetMinutes if minutes < 0 { sign = "-" minutes = -minutes } return fmt.Sprintf("%s%02d:%02d", sign, minutes/60, minutes%60) } 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)) }