From 41a65c244f0147b3f5ac64896faa9280fa540b99 Mon Sep 17 00:00:00 2001 From: zhx Date: Sat, 4 Jul 2026 00:09:06 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=96=B0=E7=9A=84=E6=88=BF?= =?UTF-8?q?=E9=97=B4=E5=BF=AB=E7=85=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/imgroup/imgroup.go | 30 +- pkg/imgroup/imgroup_test.go | 17 +- pkg/tencentim/rest_client.go | 8 +- .../internal/service/broadcast/service.go | 40 +- .../deploy/mysql/initdb/001_room_service.sql | 3 +- .../internal/storage/mysql/schema.go | 6 +- .../storage/mysql/schema_outbox_index_test.go | 39 ++ .../cmd/backfill-last7/main.go | 272 +++++++++++-- .../cmd/backfill-last7/main_test.go | 385 ++++++++++++++++++ 9 files changed, 751 insertions(+), 49 deletions(-) create mode 100644 services/room-service/internal/storage/mysql/schema_outbox_index_test.go create mode 100644 services/statistics-service/cmd/backfill-last7/main_test.go diff --git a/pkg/imgroup/imgroup.go b/pkg/imgroup/imgroup.go index 6405bcbe..6ee2d56c 100644 --- a/pkg/imgroup/imgroup.go +++ b/pkg/imgroup/imgroup.go @@ -22,8 +22,12 @@ const ( KindRegionBroadcast Kind = "region_broadcast" // globalSuffix/regionToken 是可解析协议的一部分；不要在业务代码里重复拼接这些片段。 + // globalSuffix 是 v1 全局播报群后缀：ChatRoom 类型受套餐成员上限约束（线上已触发 10038）， + // 仅保留解析与过渡期双发能力，新群一律生成 v2。 globalSuffix = "_bc_g" - regionToken = "_bc_r_" + // globalSuffixV2 是 v2 全局播报群后缀：AVChatRoom 类型，成员数不受套餐上限约束。 + globalSuffixV2 = "_bc_g_v2" + regionToken = "_bc_r_" ) // Kind 是回调分流和客户端 join 列表使用的稳定分类，不直接暴露腾讯云原始字符串判断。 @@ -37,14 +41,24 @@ type ParsedGroup struct { RoomID string } -// GlobalBroadcastGroupID 生成 App 全局播报群 ID。 +// GlobalBroadcastGroupID 生成 App 全局播报群 ID（v2，AVChatRoom）。 // app_code 先 Normalize 再进入 GroupID，确保 HTTP、gRPC、outbox 和腾讯回调看到的是同一租户标识。 func GlobalBroadcastGroupID(value string) (string, error) { return GlobalBroadcastGroupIDWithPrefix("", value) } -// GlobalBroadcastGroupIDWithPrefix 生成带环境前缀的 App 全局播报群 ID。 +// GlobalBroadcastGroupIDWithPrefix 生成带环境前缀的 App 全局播报群 ID（v2，AVChatRoom）。 func GlobalBroadcastGroupIDWithPrefix(prefix string, value string) (string, error) { + return globalBroadcastGroupIDWithSuffix(prefix, value, globalSuffixV2) +} + +// LegacyGlobalBroadcastGroupIDWithPrefix 生成 v1 全局播报群 ID。 +// 仅供过渡期双发使用：老客户端在拉取新 join 列表前仍停留在 v1 群里。 +func LegacyGlobalBroadcastGroupIDWithPrefix(prefix string, value string) (string, error) { + return globalBroadcastGroupIDWithSuffix(prefix, value, globalSuffix) +} + +func globalBroadcastGroupIDWithSuffix(prefix string, value string, suffix string) (string, error) { prefix = strings.TrimSpace(prefix) if err := validateGroupIDPrefix(prefix); err != nil { return "", err @@ -53,7 +67,7 @@ func GlobalBroadcastGroupIDWithPrefix(prefix string, value string) (string, erro if err := validateAppCodeForGroupID(app); err != nil { return "", err } - groupID := prefix + "hy_" + app + globalSuffix + groupID := prefix + "hy_" + app + suffix if len(groupID) > roomid.MaxStringIDLength { return "", fmt.Errorf("global broadcast group_id is too long") } @@ -168,6 +182,14 @@ func RoomIDFromRoomGroupID(groupID string) (string, bool) { } func parseBroadcastGroupID(groupID string) ParsedGroup { + // v2 后缀必须先于 v1 判断：v2 群和 v1 群解析为同一个 Kind，调用方不感知版本。 + if strings.HasPrefix(groupID, "hy_") && strings.HasSuffix(groupID, globalSuffixV2) { + app := strings.TrimSuffix(strings.TrimPrefix(groupID, "hy_"), globalSuffixV2) + if validateAppCodeForGroupID(app) == nil { + return ParsedGroup{Kind: KindGlobalBroadcast, AppCode: app} + } + return ParsedGroup{Kind: KindUnknown} + } if strings.HasPrefix(groupID, "hy_") && strings.HasSuffix(groupID, globalSuffix) { app := strings.TrimSuffix(strings.TrimPrefix(groupID, "hy_"), globalSuffix) if validateAppCodeForGroupID(app) == nil { diff --git a/pkg/imgroup/imgroup_test.go b/pkg/imgroup/imgroup_test.go index c8ce7911..f1ecaa08 100644 --- a/pkg/imgroup/imgroup_test.go +++ b/pkg/imgroup/imgroup_test.go @@ -7,7 +7,7 @@ func TestBroadcastGroupIDGenerateAndParse(t *testing.T) { if err != nil { t.Fatalf("GlobalBroadcastGroupID failed: %v", err) } - if global != "hy_lalu_bc_g" { + if global != "hy_lalu_bc_g_v2" { t.Fatalf("global group mismatch: %s", global) } parsedGlobal := Parse(global) @@ -15,6 +15,19 @@ func TestBroadcastGroupIDGenerateAndParse(t *testing.T) { t.Fatalf("global parse mismatch: %+v", parsedGlobal) } + // v1 旧群在过渡期仍要能解析成同一 Kind：双发与腾讯回调都依赖这一点。 + legacy, err := LegacyGlobalBroadcastGroupIDWithPrefix("", "lalu") + if err != nil { + t.Fatalf("LegacyGlobalBroadcastGroupIDWithPrefix failed: %v", err) + } + if legacy != "hy_lalu_bc_g" { + t.Fatalf("legacy global group mismatch: %s", legacy) + } + parsedLegacy := Parse(legacy) + if parsedLegacy.Kind != KindGlobalBroadcast || parsedLegacy.AppCode != "lalu" { + t.Fatalf("legacy global parse mismatch: %+v", parsedLegacy) + } + region, err := RegionBroadcastGroupID("lalu", 1001) if err != nil { t.Fatalf("RegionBroadcastGroupID failed: %v", err) @@ -63,7 +76,7 @@ func TestBroadcastGroupIDWithPrefix(t *testing.T) { if err != nil { t.Fatalf("GlobalBroadcastGroupIDWithPrefix failed: %v", err) } - if global != "test_hy_lalu_bc_g" { + if global != "test_hy_lalu_bc_g_v2" { t.Fatalf("prefixed global group mismatch: %s", global) } region, err := RegionBroadcastGroupIDWithPrefix("test_", "lalu", 1001) diff --git a/pkg/tencentim/rest_client.go b/pkg/tencentim/rest_client.go index 400bc45e..330355ab 100644 --- a/pkg/tencentim/rest_client.go +++ b/pkg/tencentim/rest_client.go @@ -21,6 +21,9 @@ const ( DefaultEndpoint = "console.tim.qq.com" // DefaultGroupType 使用 ChatRoom，保留消息历史能力；AVChatRoom 不适合需要补偿和漫游的房间消息。 DefaultGroupType = "ChatRoom" + // BroadcastGroupType 使用 AVChatRoom（直播群）：ChatRoom 受套餐成员上限约束（标准版 2000 人， + // 已在线上触发 10038 拒绝加群），AVChatRoom 成员无上限、进出免审批，播报横幅也不需要漫游补偿。 + BroadcastGroupType = "AVChatRoom" // createGroupCommand 是建群 REST 命令；房间群、全局播报群、区域播报群都复用这一个能力。 createGroupCommand = "v4/group_open_http_svc/create_group" // sendGroupMsgCommand 是服务端发群消息 REST 命令；播报只走服务端管理员账号，不开放客户端直发。 @@ -218,7 +221,10 @@ func (c *RESTClient) EnsureGroup(ctx context.Context, spec GroupSpec) error { if spec.Type == "" { spec.Type = c.cfg.GroupType } - if spec.ApplyJoinOption == "" { + if spec.Type == BroadcastGroupType { + // AVChatRoom 不支持 ApplyJoinOption，带上会被腾讯以非法参数拒绝。 + spec.ApplyJoinOption = "" + } else if spec.ApplyJoinOption == "" { spec.ApplyJoinOption = "FreeAccess" } diff --git a/services/activity-service/internal/service/broadcast/service.go b/services/activity-service/internal/service/broadcast/service.go index 53cf9a3c..2cc1a7a7 100644 --- a/services/activity-service/internal/service/broadcast/service.go +++ b/services/activity-service/internal/service/broadcast/service.go @@ -131,7 +131,7 @@ func (s *Service) EnsureBroadcastGroups(ctx context.Context) (int, error) { if err != nil { return 0, xerr.New(xerr.InvalidArgument, err.Error()) } - if err := s.publisher.EnsureGroup(ctx, broadcastGroupSpec(globalID)); err != nil { + if err := s.publisher.EnsureGroup(ctx, s.broadcastGroupSpec(globalID)); err != nil { return count, err } count++ @@ -153,7 +153,7 @@ func (s *Service) EnsureBroadcastGroups(ctx context.Context) (int, error) { if err != nil { return count, xerr.New(xerr.InvalidArgument, err.Error()) } - if err := s.publisher.EnsureGroup(ctx, broadcastGroupSpec(groupID)); err != nil { + if err := s.publisher.EnsureGroup(ctx, s.broadcastGroupSpec(groupID)); err != nil { return count, err } count++ @@ -294,10 +294,33 @@ func (s *Service) ProcessPendingBroadcasts(ctx context.Context, options WorkerOp return result, err } result.SuccessCount++ + s.dualSendLegacyGlobal(ctx, record) } return result, nil } +// dualSendLegacyGlobal 在全局播报群从 v1(ChatRoom) 切到 v2(AVChatRoom) 的过渡期内， +// 把已成功投递到 v2 群的全局播报补发一份到 v1 旧群：老客户端在重新拉取 join 列表前 +// 仍停留在旧群里。补发尽力而为，失败不影响主投递的成功状态。 +// 单个用户不会重复收到：新登录只加入 v2 群，存量会话只在 v1 群。 +func (s *Service) dualSendLegacyGlobal(ctx context.Context, record broadcastdomain.OutboxRecord) { + parsed := imgroup.ParseWithPrefix(record.GroupID, s.cfg.GroupIDPrefix) + if parsed.Kind != imgroup.KindGlobalBroadcast { + return + } + legacyID, err := imgroup.LegacyGlobalBroadcastGroupIDWithPrefix(s.cfg.GroupIDPrefix, parsed.AppCode) + if err != nil || legacyID == record.GroupID { + return + } + _ = s.publisher.PublishGroupCustomMessage(ctx, tencentim.CustomGroupMessage{ + GroupID: legacyID, + EventID: record.EventID + "_legacy", + Desc: record.BroadcastType, + Ext: "im_broadcast", + PayloadJSON: json.RawMessage(record.PayloadJSON), + }) +} + // HandleRoomEvent 把已提交的 room outbox 事实转换为服务端播报。 // 礼物、火箭和锁房只使用 room outbox 中的稳定事实，避免把展示策略侵入 Room Cell 主链路。 func (s *Service) HandleRoomEvent(ctx context.Context, envelope *roomeventsv1.EventEnvelope) (broadcastdomain.ConsumeRoomEventResult, error) { @@ -581,7 +604,18 @@ func normalizeWorkerOptions(options WorkerOptions, cfg Config) WorkerOptions { return options } -func broadcastGroupSpec(groupID string) tencentim.GroupSpec { +func (s *Service) broadcastGroupSpec(groupID string) tencentim.GroupSpec { + if imgroup.ParseWithPrefix(groupID, s.cfg.GroupIDPrefix).Kind == imgroup.KindGlobalBroadcast { + // 全局播报群使用 AVChatRoom：ChatRoom 的套餐成员上限（标准版 2000 人）曾把新设备 + // 挡在群外（10038），AVChatRoom 成员无上限、免审批，也不再需要定时清理成员。 + return tencentim.GroupSpec{ + GroupID: groupID, + Name: groupID, + Type: tencentim.BroadcastGroupType, + } + } + // 区域播报群暂保持 ChatRoom：AVChatRoom 的并发群数量受套餐配额限制，而区域群数量 = + // 品牌数 × 区域数；先由离线成员定时清理兜底，确认配额后再整体迁移。 return tencentim.GroupSpec{ GroupID: groupID, Name: groupID, diff --git a/services/room-service/deploy/mysql/initdb/001_room_service.sql b/services/room-service/deploy/mysql/initdb/001_room_service.sql index 484f2463..18b9c951 100644 --- a/services/room-service/deploy/mysql/initdb/001_room_service.sql +++ b/services/room-service/deploy/mysql/initdb/001_room_service.sql @@ -153,7 +153,8 @@ CREATE TABLE IF NOT EXISTS room_outbox ( KEY idx_room_outbox_retry (app_code, status, next_retry_at_ms, created_at_ms, event_id), KEY idx_room_outbox_claim (app_code, status, lock_until_ms, created_at_ms, event_id), KEY idx_room_outbox_event_created (app_code, event_type, created_at_ms, event_id), - KEY idx_room_outbox_retention (app_code, status, updated_at_ms, event_id) + KEY idx_room_outbox_retention (app_code, status, updated_at_ms, event_id), + KEY idx_room_outbox_room_event (app_code, room_id, event_type, created_at_ms, event_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='房间事件 outbox 表'; CREATE TABLE IF NOT EXISTS room_user_feed_entries ( diff --git a/services/room-service/internal/storage/mysql/schema.go b/services/room-service/internal/storage/mysql/schema.go index 99952b63..2659153e 100644 --- a/services/room-service/internal/storage/mysql/schema.go +++ b/services/room-service/internal/storage/mysql/schema.go @@ -144,7 +144,8 @@ func (r *Repository) Migrate(ctx context.Context) error { KEY idx_room_outbox_pending (app_code, status, next_retry_at_ms, created_at_ms, event_id), KEY idx_room_outbox_claim (app_code, status, lock_until_ms, created_at_ms, event_id), KEY idx_room_outbox_event_created (app_code, event_type, created_at_ms, event_id), - KEY idx_room_outbox_retention (app_code, status, updated_at_ms, event_id) + KEY idx_room_outbox_retention (app_code, status, updated_at_ms, event_id), + KEY idx_room_outbox_room_event (app_code, room_id, event_type, created_at_ms, event_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS room_user_feed_entries ( app_code VARCHAR(32) NOT NULL, @@ -630,6 +631,9 @@ func (r *Repository) ensureOutboxRetrySchema(ctx context.Context) error { {"idx_room_outbox_claim", `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_claim (app_code, status, lock_until_ms, created_at_ms, event_id)`}, {"idx_room_outbox_event_created", `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_event_created (app_code, event_type, created_at_ms, event_id)`}, {"idx_room_outbox_retention", `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_retention (app_code, status, updated_at_ms, event_id)`}, + // 房间维度访问路径：deleteRoomRows 的按房清理 DELETE 和按房排查事件的诊断查询都依赖它； + // 没有这个索引时两者都要扫整个 app 分片（线上 百万级 行）。 + {"idx_room_outbox_room_event", `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_room_event (app_code, room_id, event_type, created_at_ms, event_id)`}, } for _, index := range indexes { hasIndex, err := r.indexExists(ctx, "room_outbox", index.name) diff --git a/services/room-service/internal/storage/mysql/schema_outbox_index_test.go b/services/room-service/internal/storage/mysql/schema_outbox_index_test.go new file mode 100644 index 00000000..dc6bfbbe --- /dev/null +++ b/services/room-service/internal/storage/mysql/schema_outbox_index_test.go @@ -0,0 +1,39 @@ +package mysql + +import ( + "context" + "testing" + + "hyapp/internal/testutil/mysqlschema" +) + +// 已存在的旧 room_outbox 表（没有房间维度索引）必须由 Migrate 的 ensure 链补出 +// idx_room_outbox_room_event，覆盖线上 mysql_auto_migrate=true 时的升级路径。 +func TestMigrateBackfillsRoomOutboxRoomEventIndex(t *testing.T) { + schema := mysqlschema.New(t, mysqlschema.Config{ + EnvVar: "ROOM_SERVICE_MYSQL_TEST_DSN", + InitDBPath: mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), "..", "..", "..", "deploy", "mysql", "initdb", "001_room_service.sql"), + DatabasePrefix: "hyapp_room_idx_test", + }) + ctx := context.Background() + if _, err := schema.DB.ExecContext(ctx, `ALTER TABLE room_outbox DROP INDEX idx_room_outbox_room_event`); err != nil { + t.Fatalf("drop index to simulate legacy table: %v", err) + } + + repo, err := Open(ctx, schema.DSN, 4, 4) + if err != nil { + t.Fatalf("open repository: %v", err) + } + defer repo.Close() + if err := repo.Migrate(ctx); err != nil { + t.Fatalf("migrate: %v", err) + } + + hasIndex, err := repo.indexExists(ctx, "room_outbox", "idx_room_outbox_room_event") + if err != nil { + t.Fatalf("check index: %v", err) + } + if !hasIndex { + t.Fatal("Migrate did not backfill idx_room_outbox_room_event on an existing room_outbox table") + } +} diff --git a/services/statistics-service/cmd/backfill-last7/main.go b/services/statistics-service/cmd/backfill-last7/main.go index 0edd8079..3dbad996 100644 --- a/services/statistics-service/cmd/backfill-last7/main.go +++ b/services/statistics-service/cmd/backfill-last7/main.go @@ -170,7 +170,7 @@ func main() { if err := rebuildAppDayCoinMetricsFromStatistics(ctx, statsDB, state, app, statTZ, startDay, endDay); err != nil { log.Fatalf("rebuild app day coin metrics from statistics: %v", err) } - if err := rebuildSalaryTotals(ctx, walletDB, state, app, startDay, endDay, windowEndMS, loc); err != nil { + if err := rebuildSalaryTotals(ctx, walletDB, state, app, startDay, endDay, windowStartMS, windowEndMS, loc); err != nil { log.Fatalf("rebuild salary totals: %v", err) } if err := rebuildMicStats(ctx, userDB, state, app, startDay, endDay); err != nil { @@ -578,57 +578,160 @@ func rebuildCoinSellerStock(ctx context.Context, db *sql.DB, state *rebuildState } func rebuildWalletReportMetrics(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, startMS int64, endMS int64, loc *time.Location) error { - if err := rebuildCoinTotals(ctx, db, state, app, startDay, endDay, endMS, loc); err != nil { + if err := rebuildCoinTotals(ctx, db, state, app, startDay, endDay, startMS, endMS, loc); err != nil { return err } + dayEnds, err := dayEndPoints(startDay, endDay, loc) + if err != nil { + return err + } + // 按天分段扫描：每条分录的归类只依赖自身，半开区间拼接后与整窗单条大查询的结果完全一致， + // 但避免了一条 30 秒级慢 SQL 长时间持有 MVCC 读视图。 + chunkStartMS := startMS + for _, point := range dayEnds { + if err := rebuildWalletReportMetricsChunk(ctx, db, state, app, chunkStartMS, point.endMS, loc); err != nil { + return err + } + chunkStartMS = point.endMS + } + return nil +} + +type walletReportEntryRow struct { + userID int64 + counterpartyUserID int64 + assetType string + amountDelta int64 + transactionID string + createdAtMS int64 +} + +func rebuildWalletReportMetricsChunk(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS int64, endMS int64, loc *time.Location) error { + // 先只扫 wallet_entries（命中 idx_wallet_entries_asset_time），再按去重后的 transaction_id 批量取交易信息； + // 复式记账下两条分录共享一笔交易，去重能省一半 wallet_transactions 随机回表。 rows, err := db.QueryContext(ctx, ` SELECT e.user_id, e.counterparty_user_id, e.asset_type, e.available_delta + e.frozen_delta AS amount_delta, - wt.biz_type, COALESCE(JSON_UNQUOTE(JSON_EXTRACT(wt.metadata_json, '$.command_id')), wt.command_id), e.created_at_ms + e.transaction_id, e.created_at_ms FROM wallet_entries e - JOIN wallet_transactions wt ON wt.app_code = e.app_code AND wt.transaction_id = e.transaction_id WHERE e.app_code = ? AND e.asset_type IN ('COIN', 'COIN_SELLER_COIN') AND e.created_at_ms >= ? AND e.created_at_ms < ?`, app, startMS, endMS) if err != nil { return err } - defer rows.Close() + entries := []walletReportEntryRow{} + transactionIDs := map[string]struct{}{} for rows.Next() { - var userID, counterpartyUserID, amountDelta, createdAtMS int64 - var assetType, bizType, commandID string - if err := rows.Scan(&userID, &counterpartyUserID, &assetType, &amountDelta, &bizType, &commandID, &createdAtMS); err != nil { + var entry walletReportEntryRow + if err := rows.Scan(&entry.userID, &entry.counterpartyUserID, &entry.assetType, &entry.amountDelta, &entry.transactionID, &entry.createdAtMS); err != nil { + rows.Close() return err } - day := dayFromMS(createdAtMS, loc) - normalizedBiz := strings.ToLower(strings.TrimSpace(bizType)) - normalizedAsset := strings.ToUpper(strings.TrimSpace(assetType)) - if amountDelta > 0 && normalizedBiz == "salary_transfer_to_coin_seller" && normalizedAsset == "COIN_SELLER_COIN" { - dimensionUserID := userID - if counterpartyUserID > 0 { - dimensionUserID = counterpartyUserID + entries = append(entries, entry) + transactionIDs[entry.transactionID] = struct{}{} + } + if err := rows.Err(); err != nil { + rows.Close() + return err + } + if err := rows.Close(); err != nil { + return err + } + transactions, err := loadWalletReportTransactions(ctx, db, app, transactionIDs) + if err != nil { + return err + } + for _, entry := range entries { + tx, ok := transactions[entry.transactionID] + if !ok { + // 原实现是 INNER JOIN：缺交易主行的分录同样不参与口径。 + continue + } + day := dayFromMS(entry.createdAtMS, loc) + normalizedBiz := strings.ToLower(strings.TrimSpace(tx.bizType)) + normalizedAsset := strings.ToUpper(strings.TrimSpace(entry.assetType)) + if entry.amountDelta > 0 && normalizedBiz == "salary_transfer_to_coin_seller" && normalizedAsset == "COIN_SELLER_COIN" { + dimensionUserID := entry.userID + if entry.counterpartyUserID > 0 { + dimensionUserID = entry.counterpartyUserID } agg := state.ensureAgg(day, state.userDims[dimensionUserID]) - agg.salaryTransferCoin += amountDelta + agg.salaryTransferCoin += entry.amountDelta continue } if normalizedAsset != "COIN" { continue } - dim := state.userDims[userID] + dim := state.userDims[entry.userID] agg := state.ensureAgg(day, dim) switch { - case amountDelta < 0 && isConsumedCoinBizType(normalizedBiz): - agg.consumedCoin += -amountDelta - case amountDelta > 0 && isOutputCoinBizType(normalizedBiz): - agg.outputCoin += amountDelta - case amountDelta > 0 && normalizedBiz == "manual_credit": - agg.manualGrantCoin += amountDelta - case amountDelta > 0 && normalizedBiz == "resource_grant" && isPlatformGrantCommandID(commandID): - agg.platformGrantCoin += amountDelta - case amountDelta > 0 && isPlatformGrantBizType(normalizedBiz): - agg.platformGrantCoin += amountDelta + case entry.amountDelta < 0 && isConsumedCoinBizType(normalizedBiz): + agg.consumedCoin += -entry.amountDelta + case entry.amountDelta > 0 && isOutputCoinBizType(normalizedBiz): + agg.outputCoin += entry.amountDelta + case entry.amountDelta > 0 && normalizedBiz == "manual_credit": + agg.manualGrantCoin += entry.amountDelta + case entry.amountDelta > 0 && normalizedBiz == "resource_grant" && isPlatformGrantCommandID(tx.commandID): + agg.platformGrantCoin += entry.amountDelta + case entry.amountDelta > 0 && isPlatformGrantBizType(normalizedBiz): + agg.platformGrantCoin += entry.amountDelta } } - return rows.Err() + return nil +} + +type walletReportTransaction struct { + bizType string + commandID string +} + +func loadWalletReportTransactions(ctx context.Context, db *sql.DB, app string, transactionIDs map[string]struct{}) (map[string]walletReportTransaction, error) { + ids := make([]string, 0, len(transactionIDs)) + for id := range transactionIDs { + ids = append(ids, id) + } + sort.Strings(ids) + result := make(map[string]walletReportTransaction, len(ids)) + for start := 0; start < len(ids); start += 500 { + end := start + 500 + if end > len(ids) { + end = len(ids) + } + placeholders := strings.TrimRight(strings.Repeat("?,", end-start), ",") + args := []any{app} + for _, id := range ids[start:end] { + args = append(args, id) + } + // command_id 只在 resource_grant 的平台发放判定里使用，其余业务类型直接取主表列， + // 避免为每笔交易物化并解析 metadata_json。TRIM 对齐 Go 侧 TrimSpace 归一化， + // 保证 SQL 判定和消费方对 biz_type 是否为 resource_grant 的结论一致。 + rows, err := db.QueryContext(ctx, ` + SELECT transaction_id, biz_type, + CASE WHEN TRIM(biz_type) = 'resource_grant' + THEN COALESCE(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.command_id')), command_id) + ELSE command_id END + FROM wallet_transactions + WHERE app_code = ? AND transaction_id IN (`+placeholders+`)`, args...) + if err != nil { + return nil, err + } + for rows.Next() { + var id string + var tx walletReportTransaction + if err := rows.Scan(&id, &tx.bizType, &tx.commandID); err != nil { + rows.Close() + return nil, err + } + result[id] = tx + } + if err := rows.Err(); err != nil { + rows.Close() + return nil, err + } + if err := rows.Close(); err != nil { + return nil, err + } + } + return result, nil } func rebuildAppDayCoinMetricsFromStatistics(ctx context.Context, db *sql.DB, state *rebuildState, app string, statTZ string, startDay string, endDay string) error { @@ -665,17 +768,17 @@ type walletBalanceEntry struct { createdAtMS int64 } -func rebuildCoinTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowEndMS int64, loc *time.Location) error { - return rebuildWalletBalanceTotals(ctx, db, state, app, startDay, endDay, windowEndMS, loc, []string{"COIN"}, func(agg *appDayAgg, balance int64) { +func rebuildCoinTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowStartMS int64, windowEndMS int64, loc *time.Location) error { + return rebuildWalletBalanceTotals(ctx, db, state, app, startDay, endDay, windowStartMS, windowEndMS, loc, []string{"COIN"}, func(agg *appDayAgg, balance int64) { agg.coinTotal += balance }) } -func rebuildSalaryTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowEndMS int64, loc *time.Location) error { +func rebuildSalaryTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowStartMS int64, windowEndMS int64, loc *time.Location) error { for _, assetType := range []string{"HOST_SALARY_USD", "AGENCY_SALARY_USD"} { // 每个资产单独回放，可以命中 wallet_entries 的 asset/user/time 索引顺序； // 如果用户同时有多种工资钱包，也会按资产最新余额分别累计，不会因为跨资产排序打散同一用户状态。 - if err := rebuildWalletBalanceTotals(ctx, db, state, app, startDay, endDay, windowEndMS, loc, []string{assetType}, func(agg *appDayAgg, balance int64) { + if err := rebuildWalletBalanceTotals(ctx, db, state, app, startDay, endDay, windowStartMS, windowEndMS, loc, []string{assetType}, func(agg *appDayAgg, balance int64) { agg.salaryUSDMinor += balance }); err != nil { return err @@ -684,7 +787,70 @@ func rebuildSalaryTotals(ctx context.Context, db *sql.DB, state *rebuildState, a return nil } -func rebuildWalletBalanceTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowEndMS int64, loc *time.Location, assetTypes []string, applyBalance func(*appDayAgg, int64)) error { +type walletOpeningKey struct { + userID int64 + assetType string +} + +type walletOpeningRow struct { + balance int64 + createdAtMS int64 + entryID int64 +} + +// loadWalletOpeningBalances 取每个用户每种资产在 beforeMS 之前的最后一条账后余额（期初快照）。 +// 分组极值可以命中 idx_wallet_entries_asset_user_time；「最后一条」按 (created_at_ms, entry_id) +// 双键在 Go 侧判定，与全量回放 ORDER BY created_at_ms, entry_id 的口径一致，且不依赖列排序规则。 +// 返回扁平 map 而不是按用户嵌套：全量 COIN 用户的期初快照会常驻内存，嵌套 map 的每用户开销要大数倍。 +func loadWalletOpeningBalances(ctx context.Context, db *sql.DB, app string, assetTypes []string, beforeMS int64) (map[walletOpeningKey]walletOpeningRow, error) { + opening := map[walletOpeningKey]walletOpeningRow{} + if len(assetTypes) == 0 { + return opening, nil + } + placeholders := strings.TrimRight(strings.Repeat("?,", len(assetTypes)), ",") + assetArgs := make([]any, 0, len(assetTypes)) + for _, assetType := range assetTypes { + assetArgs = append(assetArgs, strings.ToUpper(strings.TrimSpace(assetType))) + } + args := []any{app} + args = append(args, assetArgs...) + args = append(args, beforeMS, app) + args = append(args, assetArgs...) + args = append(args, beforeMS) + rows, err := db.QueryContext(ctx, ` + SELECT e.user_id, e.asset_type, e.available_after + e.frozen_after AS balance_after, e.created_at_ms, e.entry_id + FROM wallet_entries e + JOIN ( + SELECT user_id, asset_type, MAX(created_at_ms) AS last_ms + FROM wallet_entries + WHERE app_code = ? AND asset_type IN (`+placeholders+`) AND created_at_ms < ? + GROUP BY user_id, asset_type + ) last ON last.user_id = e.user_id AND last.asset_type = e.asset_type AND last.last_ms = e.created_at_ms + WHERE e.app_code = ? AND e.asset_type IN (`+placeholders+`) AND e.created_at_ms < ?`, + args...) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var userID int64 + var assetType string + var candidate walletOpeningRow + if err := rows.Scan(&userID, &assetType, &candidate.balance, &candidate.createdAtMS, &candidate.entryID); err != nil { + return nil, err + } + key := walletOpeningKey{userID: userID, assetType: strings.ToUpper(strings.TrimSpace(assetType))} + if best, ok := opening[key]; ok && + (best.createdAtMS > candidate.createdAtMS || + (best.createdAtMS == candidate.createdAtMS && best.entryID >= candidate.entryID)) { + continue + } + opening[key] = candidate + } + return opening, rows.Err() +} + +func rebuildWalletBalanceTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowStartMS int64, windowEndMS int64, loc *time.Location, assetTypes []string, applyBalance func(*appDayAgg, int64)) error { dayEnds, err := dayEndPoints(startDay, endDay, loc) if err != nil { return err @@ -692,16 +858,26 @@ func rebuildWalletBalanceTotals(ctx context.Context, db *sql.DB, state *rebuildS if len(assetTypes) == 0 { return nil } + // 窗口前的历史只需要期初快照，不必整段回放；窗口内分录再按原顺序回放。 + // 任何日终点都不早于窗口起点，所以「期初快照 + 窗口回放」与全历史回放的日终余额口径完全一致。 + opening, err := loadWalletOpeningBalances(ctx, db, app, assetTypes, windowStartMS) + if err != nil { + return err + } + normalizedAssets := make([]string, 0, len(assetTypes)) + for _, assetType := range assetTypes { + normalizedAssets = append(normalizedAssets, strings.ToUpper(strings.TrimSpace(assetType))) + } placeholders := strings.TrimRight(strings.Repeat("?,", len(assetTypes)), ",") args := []any{app} - for _, assetType := range assetTypes { - args = append(args, strings.ToUpper(strings.TrimSpace(assetType))) + for _, assetType := range normalizedAssets { + args = append(args, assetType) } - args = append(args, windowEndMS) + args = append(args, windowStartMS, windowEndMS) rows, err := db.QueryContext(ctx, ` SELECT e.user_id, e.asset_type, e.available_after + e.frozen_after AS balance_after, e.created_at_ms FROM wallet_entries e - WHERE e.app_code = ? AND e.asset_type IN (`+placeholders+`) AND e.created_at_ms < ? + WHERE e.app_code = ? AND e.asset_type IN (`+placeholders+`) AND e.created_at_ms >= ? AND e.created_at_ms < ? ORDER BY e.user_id ASC, e.created_at_ms ASC, e.entry_id ASC`, args...) if err != nil { @@ -717,6 +893,13 @@ func rebuildWalletBalanceTotals(ctx context.Context, db *sql.DB, state *rebuildS dim := state.userDims[currentUserID] index := 0 balancesByAsset := map[string]int64{} + for _, asset := range normalizedAssets { + key := walletOpeningKey{userID: currentUserID, assetType: asset} + if row, ok := opening[key]; ok { + balancesByAsset[asset] = row.balance + delete(opening, key) + } + } for _, point := range dayEnds { for index < len(entries) && entries[index].createdAtMS < point.endMS { balancesByAsset[entries[index].assetType] = entries[index].balance @@ -751,6 +934,21 @@ func rebuildWalletBalanceTotals(ctx context.Context, db *sql.DB, state *rebuildS return err } flush() + // 窗口内没有新分录的用户不会出现在流式结果里，但他们的期初存量仍然计入每一个日终总额， + // 与全历史回放（旧分录在第一个日终点前就被消费掉）的结果一致。 + leftoverTotals := map[int64]int64{} + for key, row := range opening { + if key.userID <= 0 { + continue + } + leftoverTotals[key.userID] += row.balance + } + for userID, balanceTotal := range leftoverTotals { + dim := state.userDims[userID] + for _, point := range dayEnds { + applyBalance(state.ensureAgg(point.day, dim), balanceTotal) + } + } return nil } diff --git a/services/statistics-service/cmd/backfill-last7/main_test.go b/services/statistics-service/cmd/backfill-last7/main_test.go new file mode 100644 index 00000000..99e9e136 --- /dev/null +++ b/services/statistics-service/cmd/backfill-last7/main_test.go @@ -0,0 +1,385 @@ +package main + +import ( + "context" + "database/sql" + "fmt" + "strings" + "testing" + "time" + + "hyapp/internal/testutil/mysqlschema" +) + +// 这些测试验证钱包扫描的性能改写与旧实现口径逐字段一致： +// rebuildWalletBalanceTotals 的「期初快照 + 窗口回放」必须等价于旧的全历史回放， +// rebuildWalletReportMetrics 的「按天分段 + 批量取交易」必须等价于旧的整窗 JOIN。 +// 基准实现是改写前代码的逐字拷贝，不要随新实现同步修改。 + +const ( + testApp = "lalu" + testStartDay = "2026-06-29" + testEndDay = "2026-07-01" +) + +func newWalletSchema(t *testing.T) *mysqlschema.Schema { + t.Helper() + return mysqlschema.New(t, mysqlschema.Config{ + InitDBPath: mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), + "..", "..", "..", "wallet-service", "deploy", "mysql", "initdb", "001_wallet_service.sql"), + DatabasePrefix: "hy_test_backfill", + }) +} + +func testWindow(t *testing.T) (loc *time.Location, windowStartMS int64, windowEndMS int64) { + t.Helper() + loc = time.UTC + start, err := parseDay(testStartDay, loc) + if err != nil { + t.Fatal(err) + } + end, err := parseDay(testEndDay, loc) + if err != nil { + t.Fatal(err) + } + return loc, start.UTC().UnixMilli(), end.AddDate(0, 0, 1).UTC().UnixMilli() +} + +type seedEntry struct { + entryID int64 + app string + txID string + userID int64 + assetType string + availableDelta int64 + frozenDelta int64 + availableAfter int64 + frozenAfter int64 + counterparty int64 + createdAtMS int64 +} + +type seedTransaction struct { + app string + txID string + commandID string + bizType string + metadataJSON sql.NullString +} + +func seedWalletData(t *testing.T, db *sql.DB, transactions []seedTransaction, entries []seedEntry) { + t.Helper() + ctx := context.Background() + for _, tx := range transactions { + if _, err := db.ExecContext(ctx, ` + INSERT INTO wallet_transactions (app_code, transaction_id, command_id, biz_type, status, request_hash, metadata_json, created_at_ms, updated_at_ms) + VALUES (?, ?, ?, ?, 'committed', ?, ?, 1, 1)`, + tx.app, tx.txID, tx.commandID, tx.bizType, "hash_"+tx.txID, tx.metadataJSON); err != nil { + t.Fatalf("seed transaction %s: %v", tx.txID, err) + } + } + for _, entry := range entries { + if _, err := db.ExecContext(ctx, ` + INSERT INTO wallet_entries (entry_id, app_code, transaction_id, user_id, asset_type, available_delta, frozen_delta, available_after, frozen_after, counterparty_user_id, room_id, created_at_ms) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '', ?)`, + entry.entryID, entry.app, entry.txID, entry.userID, entry.assetType, + entry.availableDelta, entry.frozenDelta, entry.availableAfter, entry.frozenAfter, + entry.counterparty, entry.createdAtMS); err != nil { + t.Fatalf("seed entry %d: %v", entry.entryID, err) + } + } +} + +func newTestState() *rebuildState { + state := &rebuildState{ + appDays: map[appDayKey]*appDayAgg{}, + registeredByUserDay: map[string]struct{}{}, + payersByKey: map[string]payerRow{}, + activeByKey: map[string]activeRow{}, + micByKey: map[string]micRow{}, + userIDs: map[int64]struct{}{}, + userDims: map[int64]countryRegion{}, + knownUsers: map[int64]struct{}{}, + } + // 一部分用户带维度，一部分故意缺失（走零值维度分支）。 + for userID := int64(101); userID <= 110; userID++ { + state.userDims[userID] = countryRegion{countryID: userID % 5, regionID: userID % 3} + } + state.userDims[202] = countryRegion{countryID: 9, regionID: 2} + return state +} + +func requireEqualAppDays(t *testing.T, got map[appDayKey]*appDayAgg, want map[appDayKey]*appDayAgg) { + t.Helper() + for key := range want { + if got[key] == nil { + t.Errorf("missing agg for %+v (reference has %+v)", key, *want[key]) + } + } + for key, gotAgg := range got { + wantAgg := want[key] + if wantAgg == nil { + t.Errorf("unexpected agg for %+v: %+v", key, *gotAgg) + continue + } + if *gotAgg != *wantAgg { + t.Errorf("agg mismatch for %+v:\n got=%+v\nwant=%+v", key, *gotAgg, *wantAgg) + } + } +} + +// referenceRebuildWalletBalanceTotals 是优化前的全历史回放实现，逐字拷贝作为口径基准。 +func referenceRebuildWalletBalanceTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowEndMS int64, loc *time.Location, assetTypes []string, applyBalance func(*appDayAgg, int64)) error { + dayEnds, err := dayEndPoints(startDay, endDay, loc) + if err != nil { + return err + } + if len(assetTypes) == 0 { + return nil + } + placeholders := strings.TrimRight(strings.Repeat("?,", len(assetTypes)), ",") + args := []any{app} + for _, assetType := range assetTypes { + args = append(args, strings.ToUpper(strings.TrimSpace(assetType))) + } + args = append(args, windowEndMS) + rows, err := db.QueryContext(ctx, ` + SELECT e.user_id, e.asset_type, e.available_after + e.frozen_after AS balance_after, e.created_at_ms + FROM wallet_entries e + WHERE e.app_code = ? AND e.asset_type IN (`+placeholders+`) AND e.created_at_ms < ? + ORDER BY e.user_id ASC, e.created_at_ms ASC, e.entry_id ASC`, + args...) + if err != nil { + return err + } + defer rows.Close() + var currentUserID int64 + entries := []walletBalanceEntry{} + flush := func() { + if currentUserID <= 0 || len(entries) == 0 { + return + } + dim := state.userDims[currentUserID] + index := 0 + balancesByAsset := map[string]int64{} + for _, point := range dayEnds { + for index < len(entries) && entries[index].createdAtMS < point.endMS { + balancesByAsset[entries[index].assetType] = entries[index].balance + index++ + } + if len(balancesByAsset) == 0 { + continue + } + balanceTotal := int64(0) + for _, balance := range balancesByAsset { + balanceTotal += balance + } + applyBalance(state.ensureAgg(point.day, dim), balanceTotal) + } + } + for rows.Next() { + var entry walletBalanceEntry + if err := rows.Scan(&entry.userID, &entry.assetType, &entry.balance, &entry.createdAtMS); err != nil { + return err + } + entry.assetType = strings.ToUpper(strings.TrimSpace(entry.assetType)) + if currentUserID != 0 && entry.userID != currentUserID { + flush() + entries = entries[:0] + } + currentUserID = entry.userID + entries = append(entries, entry) + } + if err := rows.Err(); err != nil { + return err + } + flush() + return nil +} + +// referenceRebuildWalletReportMetrics 是优化前的整窗 JOIN 实现，逐字拷贝作为口径基准。 +func referenceRebuildWalletReportMetrics(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, startMS int64, endMS int64, loc *time.Location) error { + if err := referenceRebuildWalletBalanceTotals(ctx, db, state, app, startDay, endDay, endMS, loc, []string{"COIN"}, func(agg *appDayAgg, balance int64) { + agg.coinTotal += balance + }); err != nil { + return err + } + rows, err := db.QueryContext(ctx, ` + SELECT e.user_id, e.counterparty_user_id, e.asset_type, e.available_delta + e.frozen_delta AS amount_delta, + wt.biz_type, COALESCE(JSON_UNQUOTE(JSON_EXTRACT(wt.metadata_json, '$.command_id')), wt.command_id), e.created_at_ms + FROM wallet_entries e + JOIN wallet_transactions wt ON wt.app_code = e.app_code AND wt.transaction_id = e.transaction_id + WHERE e.app_code = ? AND e.asset_type IN ('COIN', 'COIN_SELLER_COIN') AND e.created_at_ms >= ? AND e.created_at_ms < ?`, + app, startMS, endMS) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + var userID, counterpartyUserID, amountDelta, createdAtMS int64 + var assetType, bizType, commandID string + if err := rows.Scan(&userID, &counterpartyUserID, &assetType, &amountDelta, &bizType, &commandID, &createdAtMS); err != nil { + return err + } + day := dayFromMS(createdAtMS, loc) + normalizedBiz := strings.ToLower(strings.TrimSpace(bizType)) + normalizedAsset := strings.ToUpper(strings.TrimSpace(assetType)) + if amountDelta > 0 && normalizedBiz == "salary_transfer_to_coin_seller" && normalizedAsset == "COIN_SELLER_COIN" { + dimensionUserID := userID + if counterpartyUserID > 0 { + dimensionUserID = counterpartyUserID + } + agg := state.ensureAgg(day, state.userDims[dimensionUserID]) + agg.salaryTransferCoin += amountDelta + continue + } + if normalizedAsset != "COIN" { + continue + } + dim := state.userDims[userID] + agg := state.ensureAgg(day, dim) + switch { + case amountDelta < 0 && isConsumedCoinBizType(normalizedBiz): + agg.consumedCoin += -amountDelta + case amountDelta > 0 && isOutputCoinBizType(normalizedBiz): + agg.outputCoin += amountDelta + case amountDelta > 0 && normalizedBiz == "manual_credit": + agg.manualGrantCoin += amountDelta + case amountDelta > 0 && normalizedBiz == "resource_grant" && isPlatformGrantCommandID(commandID): + agg.platformGrantCoin += amountDelta + case amountDelta > 0 && isPlatformGrantBizType(normalizedBiz): + agg.platformGrantCoin += amountDelta + } + } + return rows.Err() +} + +func jsonMeta(value string) sql.NullString { + return sql.NullString{String: value, Valid: true} +} + +// seedBalanceAndReportFixture 铺一份同时覆盖余额回放和报表归类所有分支的数据。 +func seedBalanceAndReportFixture(t *testing.T, db *sql.DB, windowStartMS int64, windowEndMS int64) { + t.Helper() + day1 := windowStartMS // 2026-06-29 00:00 UTC + day2 := day1 + 24*3600*1000 // 2026-06-30 00:00 UTC + day3 := day2 + 24*3600*1000 // 2026-07-01 00:00 UTC + pre := windowStartMS - 10*24*3600*1000 // 窗口前 10 天 + + transactions := []seedTransaction{ + {testApp, "tx_pre_1", "cmd_pre_1", "recharge_credit", sql.NullString{}}, + {testApp, "tx_pre_2", "cmd_pre_2", "recharge_credit", sql.NullString{}}, + {testApp, "tx_pre_3", "cmd_pre_3", "recharge_credit", sql.NullString{}}, + {testApp, "tx_pre_4", "cmd_pre_4", "recharge_credit", sql.NullString{}}, + {testApp, "tx_pre_5", "cmd_pre_5", "recharge_credit", sql.NullString{}}, + {testApp, "tx_gift_neg", "cmd_gift_neg", "direct_gift_debit", sql.NullString{}}, + {testApp, "tx_gift_pos", "cmd_gift_pos", "gift_debit", sql.NullString{}}, + {testApp, "tx_gift_direct_pos", "cmd_gift_direct_pos", "direct_gift_debit", sql.NullString{}}, + {testApp, "tx_manual", "cmd_manual", "manual_credit", sql.NullString{}}, + {testApp, "tx_grant_meta", "cmd_grant_meta", "resource_grant", jsonMeta(`{"command_id":"weekly_star:abc"}`)}, + {testApp, "tx_grant_nometa", "wtask_grant_2", "resource_grant", jsonMeta(`{"other":1}`)}, + {testApp, "tx_grant_other", "cmd_grant_other", "resource_grant", sql.NullString{}}, + {testApp, "tx_grant_case", "cp_weekly_rank:z", "Resource_Grant", sql.NullString{}}, + {testApp, "tx_grant_leading", "cmd_not_platform", " resource_grant", jsonMeta(`{"command_id":"weekly_star:lead"}`)}, + {testApp, "tx_task", "cmd_task", "task_reward", sql.NullString{}}, + {testApp, "tx_redpacket", "cmd_redpacket", "red_packet_create", sql.NullString{}}, + {testApp, "tx_wheel", "cmd_wheel", "wheel_draw_debit", sql.NullString{}}, + {testApp, "tx_claim", "cmd_claim", "red_packet_claim", sql.NullString{}}, + {testApp, "tx_salary_transfer", "cmd_salary_transfer", "salary_transfer_to_coin_seller", jsonMeta(`{"command_id":"ignored"}`)}, + {testApp, "tx_seller_manual", "cmd_seller_manual", "manual_credit", sql.NullString{}}, + {testApp, "tx_lower", "cmd_lower", "manual_credit", sql.NullString{}}, + {testApp, "tx_boundary", "cmd_boundary", "manual_credit", sql.NullString{}}, + // 故意不给 tx_missing 建交易主行：INNER JOIN 语义下该分录不参与口径。 + {"other", "tx_other_app", "cmd_other_app", "manual_credit", sql.NullString{}}, + } + + entries := []seedEntry{ + // user 101：只有窗口前分录（期初快照兜底路径），日终余额恒为 500。 + {1, testApp, "tx_pre_1", 101, "COIN", 300, 0, 300, 0, 0, pre}, + {2, testApp, "tx_pre_2", 101, "COIN", 200, 0, 400, 100, 0, pre + 1000}, + // user 104：窗口前同一毫秒两条分录，entry_id 大者（余额 777）才是口径上的期初。 + {3, testApp, "tx_pre_3", 104, "COIN", 100, 0, 111, 0, 0, pre + 5000}, + {4, testApp, "tx_pre_4", 104, "COIN", 100, 0, 777, 0, 0, pre + 5000}, + // user 102：窗口前期初 900，day2 内变为 250。 + {5, testApp, "tx_pre_5", 102, "COIN", 900, 0, 900, 0, 0, pre + 9000}, + {6, testApp, "tx_gift_neg", 102, "COIN", -100, 0, 250, 0, 0, day2 + 3600_000}, + // user 103：只有窗口内分录，day3 起才计入。 + {7, testApp, "tx_gift_pos", 103, "COIN", 80, 0, 80, 0, 0, day3 + 7200_000}, + // user 105：COIN 与 HOST_SALARY_USD 双资产。 + {8, testApp, "tx_manual", 105, "COIN", 50, 0, 50, 0, 0, day1 + 1000}, + {9, testApp, "tx_grant_meta", 105, "HOST_SALARY_USD", 30, 0, 30, 0, 0, pre + 100}, + // user 106：负余额也按原样累计。 + {10, testApp, "tx_grant_nometa", 106, "COIN", 30, 0, -20, 0, 0, day1 + 2000}, + // user 107：AGENCY_SALARY_USD 只在窗口内。 + {11, testApp, "tx_grant_other", 107, "AGENCY_SALARY_USD", 12, 0, 12, 0, 0, day2 + 100}, + // user 108：HOST_SALARY_USD 只有期初。 + {12, testApp, "tx_task", 108, "HOST_SALARY_USD", 40, 0, 40, 0, 0, pre + 200}, + // 报表归类分支（均在窗口内，含各 biz_type 与正负号组合）。 + {13, testApp, "tx_grant_case", 105, "COIN", 5, 0, 55, 0, 0, day2 + 5000}, + {14, testApp, "tx_redpacket", 102, "COIN", -40, 0, 210, 0, 0, day2 + 6000}, + {15, testApp, "tx_wheel", 102, "COIN", -60, 0, 150, 0, 0, day3 + 1000}, + {16, testApp, "tx_claim", 103, "COIN", 25, 0, 105, 0, 0, day3 + 8000}, + {17, testApp, "tx_salary_transfer", 201, "COIN_SELLER_COIN", 500, 0, 500, 0, 202, day2 + 7000}, + {18, testApp, "tx_seller_manual", 201, "COIN_SELLER_COIN", 10, 0, 510, 0, 0, day2 + 8000}, + {19, testApp, "tx_missing", 109, "COIN", 33, 0, 33, 0, 0, day2 + 9000}, + // 小写 asset_type：ci 排序规则下新旧实现都会命中并归一化成大写。 + {20, testApp, "tx_lower", 110, "coin", 15, 0, 15, 0, 0, day1 + 3000}, + // 边界：恰好等于窗口起点与 day2 起点的分录。 + {21, testApp, "tx_boundary", 106, "COIN", 6, 0, -14, 0, 0, day2}, + {22, testApp, "tx_grant_meta", 104, "COIN", 30, 0, 807, 0, 0, windowStartMS}, + // 其他 app 的分录必须被过滤。 + {23, "other", "tx_other_app", 101, "COIN", 999, 0, 999, 0, 0, day1 + 100}, + // 正向 direct_gift_debit 归产出（新旧一致）。 + {24, testApp, "tx_gift_direct_pos", 103, "COIN", 10, 0, 90, 0, 0, day3 + 7500}, + // 前导空格 biz_type：Go 侧 TrimSpace 归到 resource_grant，SQL 侧 TRIM 门必须同样命中 metadata command_id。 + {25, testApp, "tx_grant_leading", 105, "COIN", 7, 0, 62, 0, 0, day2 + 9500}, + } + seedWalletData(t, db, transactions, entries) +} + +func TestRebuildWalletBalanceTotalsMatchesFullReplay(t *testing.T) { + schema := newWalletSchema(t) + ctx := context.Background() + loc, windowStartMS, windowEndMS := testWindow(t) + seedBalanceAndReportFixture(t, schema.DB, windowStartMS, windowEndMS) + + // 多资产切片没有生产调用方，但要覆盖占位符拼装的通用分支。 + for _, assets := range [][]string{{"COIN"}, {"HOST_SALARY_USD"}, {"AGENCY_SALARY_USD"}, {"HOST_SALARY_USD", "AGENCY_SALARY_USD"}} { + newState := newTestState() + refState := newTestState() + apply := func(agg *appDayAgg, balance int64) { agg.coinTotal += balance } + if err := rebuildWalletBalanceTotals(ctx, schema.DB, newState, testApp, testStartDay, testEndDay, windowStartMS, windowEndMS, loc, assets, apply); err != nil { + t.Fatalf("new impl (%v): %v", assets, err) + } + if err := referenceRebuildWalletBalanceTotals(ctx, schema.DB, refState, testApp, testStartDay, testEndDay, windowEndMS, loc, assets, apply); err != nil { + t.Fatalf("reference impl (%v): %v", assets, err) + } + t.Run(fmt.Sprintf("assets_%s", strings.Join(assets, "_")), func(t *testing.T) { + requireEqualAppDays(t, newState.appDays, refState.appDays) + }) + } +} + +func TestRebuildWalletReportMetricsMatchesJoinQuery(t *testing.T) { + schema := newWalletSchema(t) + ctx := context.Background() + loc, windowStartMS, windowEndMS := testWindow(t) + seedBalanceAndReportFixture(t, schema.DB, windowStartMS, windowEndMS) + + newState := newTestState() + refState := newTestState() + if err := rebuildWalletReportMetrics(ctx, schema.DB, newState, testApp, testStartDay, testEndDay, windowStartMS, windowEndMS, loc); err != nil { + t.Fatalf("new impl: %v", err) + } + if err := referenceRebuildWalletReportMetrics(ctx, schema.DB, refState, testApp, testStartDay, testEndDay, windowStartMS, windowEndMS, loc); err != nil { + t.Fatalf("reference impl: %v", err) + } + requireEqualAppDays(t, newState.appDays, refState.appDays) + + // 抽查一个绝对值，防止两套实现同时错成一样：user 202（维度 9/2）在 day2 收到 500 工资转账。 + key := appDayKey{day: "2026-06-30", countryID: 9, regionID: 2} + agg := newState.appDays[key] + if agg == nil || agg.salaryTransferCoin != 500 { + t.Errorf("expected salaryTransferCoin=500 for %+v, got %+v", key, agg) + } +}