386 lines
16 KiB
Go
386 lines
16 KiB
Go
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)
|
||
}
|
||
}
|