增加新的房间快照
This commit is contained in:
zhx
parent
b83c3f38d4
commit
41a65c244f
@ -22,7 +22,11 @@ const (
|
||||
KindRegionBroadcast Kind = "region_broadcast"
|
||||
|
||||
// globalSuffix/regionToken 是可解析协议的一部分；不要在业务代码里重复拼接这些片段。
|
||||
// globalSuffix 是 v1 全局播报群后缀：ChatRoom 类型受套餐成员上限约束（线上已触发 10038），
|
||||
// 仅保留解析与过渡期双发能力，新群一律生成 v2。
|
||||
globalSuffix = "_bc_g"
|
||||
// globalSuffixV2 是 v2 全局播报群后缀：AVChatRoom 类型，成员数不受套餐上限约束。
|
||||
globalSuffixV2 = "_bc_g_v2"
|
||||
regionToken = "_bc_r_"
|
||||
)
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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"
|
||||
}
|
||||
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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 (
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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")
|
||||
}
|
||||
}
|
||||
@ -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
|
||||
}
|
||||
|
||||
|
||||
385
services/statistics-service/cmd/backfill-last7/main_test.go
Normal file
385
services/statistics-service/cmd/backfill-last7/main_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user