From d8c07b07acf7a30d5cb6b19c38e8cde8b25b2cbe Mon Sep 17 00:00:00 2001 From: zhx Date: Fri, 3 Jul 2026 14:54:21 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E4=BA=86=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../modules/payment/legacy_bill_source.go | 143 ++++++++++++++---- .../payment/legacy_bill_source_test.go | 41 ++++- 2 files changed, 149 insertions(+), 35 deletions(-) diff --git a/server/admin/internal/modules/payment/legacy_bill_source.go b/server/admin/internal/modules/payment/legacy_bill_source.go index 50440744..6b95ac58 100644 --- a/server/admin/internal/modules/payment/legacy_bill_source.go +++ b/server/admin/internal/modules/payment/legacy_bill_source.go @@ -109,17 +109,13 @@ func (s *MongoRechargeBillSource) ListRechargeBills(ctx context.Context, query l if s == nil || s.collection == nil { return nil, 0, fmt.Errorf("legacy bill source is not configured") } - filter, matchable, err := s.billFilter(ctx, query) + filter, regionCodes, matchable, err := s.billFilter(ctx, query) if err != nil { return nil, 0, err } if !matchable { return []rechargeBillDTO{}, 0, nil } - 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) - } page := query.Page if page < 1 { page = 1 @@ -128,6 +124,17 @@ func (s *MongoRechargeBillSource) ListRechargeBills(ctx context.Context, query l 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)). @@ -136,26 +143,77 @@ func (s *MongoRechargeBillSource) ListRechargeBills(ctx context.Context, query l 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, 0, fmt.Errorf("decode legacy recharge bill for %s: %w", s.config.AppCode, err) + 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, 0, fmt.Errorf("iterate legacy recharge bills for %s: %w", s.config.AppCode, err) + return nil, fmt.Errorf("iterate legacy recharge bills for %s: %w", s.config.AppCode, err) } - return items, total, nil + 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, matchable, err := s.billFilter(ctx, query) + filter, regionCodes, matchable, err := s.billFilter(ctx, query) if err != nil { return rechargeBillSummaryDTO{}, err } @@ -174,19 +232,18 @@ func (s *MongoRechargeBillSource) SummarizeRechargeBills(ctx context.Context, qu bson.M{"$convert": bson.M{"input": "$$this.content", "to": "long", "onError": 0, "onNull": 0}}, }}, }} - cursor, err := s.collection.Aggregate(ctx, mongo.Pipeline{ - {{Key: "$match", Value: filter}}, - {{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}, - }}}, - }) + 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) } @@ -228,7 +285,8 @@ func (s *MongoRechargeBillSource) SummarizeRechargeBills(ctx context.Context, qu } // billFilter 构造与 hyapp 账单列表同口径的 Mongo 过滤条件；返回 matchable=false 表示筛选注定无结果。 -func (s *MongoRechargeBillSource) billFilter(ctx context.Context, query legacyRechargeBillQuery) (bson.M, bool, error) { +// 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", @@ -255,14 +313,45 @@ func (s *MongoRechargeBillSource) billFilter(ctx context.Context, query legacyRe if query.RegionID > 0 { codes, handled, err := s.regionCountryCodes(ctx, query.RegionID) if err != nil { - return nil, false, err + return nil, nil, false, err } if !handled || len(codes) == 0 { - return nil, false, nil + return nil, nil, false, nil } - filter["countryCode"] = bson.M{"$in": codes} + return filter, codes, true, nil } - return filter, true, nil + return filter, nil, true, nil +} + +const legacyUserProfileCollection = "user_run_profile" + +// legacyRegionLookupStages 按付款用户的 user_run_profile.countryCode 过滤区域；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{"$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) { diff --git a/server/admin/internal/modules/payment/legacy_bill_source_test.go b/server/admin/internal/modules/payment/legacy_bill_source_test.go index 8383f322..c07c9c3d 100644 --- a/server/admin/internal/modules/payment/legacy_bill_source_test.go +++ b/server/admin/internal/modules/payment/legacy_bill_source_test.go @@ -124,7 +124,7 @@ func TestLegacyBillFilterBuildsQuery(t *testing.T) { config: yumiBillSourceConfig(), regionResolvers: []legacyRegionCountryResolver{staticLegacyRegionResolver{codes: []string{"tr", "SA"}, handled: true}}, } - filter, matchable, err := source.billFilter(context.Background(), legacyRechargeBillQuery{ + filter, regionCodes, matchable, err := source.billFilter(context.Background(), legacyRechargeBillQuery{ Keyword: "GPA.1", RegionID: 9_000_000_000_000_000_001, StartAtMS: 1_000, @@ -140,13 +140,12 @@ func TestLegacyBillFilterBuildsQuery(t *testing.T) { if !ok || !timeRange["$gte"].(time.Time).Equal(time.UnixMilli(1_000)) || !timeRange["$lt"].(time.Time).Equal(time.UnixMilli(2_000)) { t.Fatalf("unexpected time range: %v", filter["createTime"]) } - countries, ok := filter["countryCode"].(bson.M) - if !ok { - t.Fatalf("country filter missing: %v", filter) + // likei 账单不带国家码；区域条件必须以国家码清单返回，由聚合阶段按付款用户资料匹配。 + if len(regionCodes) != 2 || regionCodes[0] != "TR" || regionCodes[1] != "SA" { + t.Fatalf("unexpected region codes: %v", regionCodes) } - codes, ok := countries["$in"].([]string) - if !ok || len(codes) != 2 || codes[0] != "TR" || codes[1] != "SA" { - t.Fatalf("unexpected country codes: %v", countries["$in"]) + if _, ok := filter["countryCode"]; ok { + t.Fatalf("bill countryCode filter should not be used: %v", filter) } if _, ok := filter["$or"]; !ok { t.Fatalf("keyword filter missing: %v", filter) @@ -155,7 +154,7 @@ func TestLegacyBillFilterBuildsQuery(t *testing.T) { func TestLegacyBillFilterUnresolvedRegionMatchesNothing(t *testing.T) { source := &MongoRechargeBillSource{config: yumiBillSourceConfig()} - _, matchable, err := source.billFilter(context.Background(), legacyRechargeBillQuery{RegionID: 123}) + _, _, matchable, err := source.billFilter(context.Background(), legacyRechargeBillQuery{RegionID: 123}) if err != nil { t.Fatalf("billFilter err=%v", err) } @@ -163,3 +162,29 @@ func TestLegacyBillFilterUnresolvedRegionMatchesNothing(t *testing.T) { t.Fatalf("unresolved region should not be matchable") } } + +func TestLegacyRegionLookupStages(t *testing.T) { + if stages := legacyRegionLookupStages(nil); len(stages) != 0 { + t.Fatalf("empty codes should not add stages: %v", stages) + } + stages := legacyRegionLookupStages([]string{"TR", "SA"}) + if len(stages) != 3 { + t.Fatalf("expected lookup/addFields/match stages, got %d", len(stages)) + } + lookup, ok := stages[0][0].Value.(bson.M) + if !ok || lookup["from"] != legacyUserProfileCollection || lookup["localField"] != "acceptUserId" || lookup["foreignField"] != "_id" { + t.Fatalf("unexpected lookup stage: %v", stages[0]) + } + match, ok := stages[2][0].Value.(bson.M) + if !ok { + t.Fatalf("unexpected match stage: %v", stages[2]) + } + condition, ok := match["legacyBillCountry"].(bson.M) + if !ok { + t.Fatalf("match stage missing country condition: %v", match) + } + codes, ok := condition["$in"].([]string) + if !ok || len(codes) != 2 || codes[0] != "TR" || codes[1] != "SA" { + t.Fatalf("unexpected region codes in match: %v", condition["$in"]) + } +}