admin/dashboard-cdc-worker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: sync gift reconcile to daily metrics

Browse Source
This commit is contained in:
zhx 2026-05-28 12:08:22 +08:00
parent b2d351d489
commit 24893460bc
2 changed files with 89 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
internal/storage/gift_sql.go
View File

@ -73,15 +73,35 @@ WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origi
return err return err
} }
if _, err := tx.ExecContext(ctx, ` if _, err := tx.ExecContext(ctx, `
UPDATE country_dashboard_daily_metric
SET gift_consume = 0,
lucky_gift_total_flow = 0,
lucky_gift_user = 0,
lucky_gift_payout = 0,
lucky_gift_anchor_share = 0,
refreshed_at = NOW()
WHERE stat_date = ? AND sys_origin = ?`,
dayKey, sysOrigin); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
DELETE FROM country_dashboard_period_user_metric DELETE FROM country_dashboard_period_user_metric
WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origin = ? WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origin = ?
AND metric_type = ?`, dayKey, statTimezone, sysOrigin, dashboard.MetricLuckyGiftUser); err != nil { AND metric_type = ?`, dayKey, statTimezone, sysOrigin, dashboard.MetricLuckyGiftUser); err != nil {
return err return err
} }
if _, err := tx.ExecContext(ctx, `
DELETE FROM country_dashboard_daily_user_metric
WHERE stat_date = ? AND sys_origin = ? AND metric_type = ?`, dayKey, sysOrigin, dashboard.MetricLuckyGiftUser); err != nil {
return err
}
} }
if err := r.upsertGiftMetricRows(ctx, tx, statTimezone, dayStart, dayEnd, rows); err != nil { if err := r.upsertGiftMetricRows(ctx, tx, statTimezone, dayStart, dayEnd, rows); err != nil {
return err return err
} }
if err := r.upsertDailyGiftMetricRows(ctx, tx, dayStart, rows); err != nil {
return err
}
} }
return nil return nil
} }
@ -153,6 +173,67 @@ ON DUPLICATE KEY UPDATE
return nil return nil
} }
func (r *MySQLRepository) upsertDailyGiftMetricRows(ctx context.Context, tx *sql.Tx,
dayStart time.Time, rows map[string]*giftMetricRow) error {
dayKey := dayStart.Format("2006-01-02")
var metricRows []*giftMetricRow
for _, row := range rows {
if row.day != dayKey || row.empty() {
continue
}
if row.sysOrigin == "" {
row.sysOrigin = defaultSysOrigin(r.cfg.Dashboard.SysOrigins)
}
if !r.allowedOrigin(row.sysOrigin) {
continue
}
if row.country.Code == "" {
row.country.Code = r.cfg.Dashboard.UnknownCountryCode
}
if row.country.Name == "" {
row.country.Name = row.country.Code
}
metricRows = append(metricRows, row)
}
if len(metricRows) == 0 {
return nil
}
args := make([]interface{}, 0, len(metricRows)*10)
for _, row := range metricRows {
args = append(args, dayKey, dateNumber(dayStart), row.sysOrigin, row.country.Code, row.country.Name,
row.giftConsume, row.luckyGiftTotalFlow, len(row.luckyGiftUsers), row.luckyGiftPayout,
row.luckyGiftAnchorShare)
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO country_dashboard_daily_metric (
stat_date, date_number, sys_origin, country_code, country_name, gift_consume,
lucky_gift_total_flow, lucky_gift_user, lucky_gift_payout, lucky_gift_anchor_share
) VALUES `+placeholders(len(metricRows), 10)+`
ON DUPLICATE KEY UPDATE
country_name = VALUES(country_name),
gift_consume = VALUES(gift_consume),
lucky_gift_total_flow = VALUES(lucky_gift_total_flow),
lucky_gift_user = VALUES(lucky_gift_user),
lucky_gift_payout = VALUES(lucky_gift_payout),
lucky_gift_anchor_share = VALUES(lucky_gift_anchor_share),
refreshed_at = NOW()`, args...); err != nil {
return err
}
for _, row := range metricRows {
if len(row.luckyGiftUsers) == 0 {
continue
}
key := dailyUserKey{
day: dayStart, sysOrigin: row.sysOrigin, countryCode: row.country.Code,
countryName: row.country.Name, metricType: dashboard.MetricLuckyGiftUser,
}
if _, err := bulkInsertDailyUsers(ctx, tx, key, int64SetToSlice(row.luckyGiftUsers)); err != nil {
return err
}
}
return nil
}
func (r *MySQLRepository) loadSourceUserCountriesByID(ctx context.Context, ids []int64) (map[int64]dashboard.UserCountry, error) { func (r *MySQLRepository) loadSourceUserCountriesByID(ctx context.Context, ids []int64) (map[int64]dashboard.UserCountry, error) {
result := make(map[int64]dashboard.UserCountry) result := make(map[int64]dashboard.UserCountry)
for start := 0; start < len(ids); start += bulkInsertChunkSize { for start := 0; start < len(ids); start += bulkInsertChunkSize {

8
internal/storage/schema.go
View File

@ -164,6 +164,14 @@ func (r *MySQLRepository) ensureMetricColumns(ctx context.Context) error {
"ALTER TABLE country_dashboard_period_metric ADD COLUMN daily_recharge_user bigint NOT NULL DEFAULT 0 AFTER game_total_flow"); err != nil { "ALTER TABLE country_dashboard_period_metric ADD COLUMN daily_recharge_user bigint NOT NULL DEFAULT 0 AFTER game_total_flow"); err != nil {
return err return err
} }
if err := r.addColumnIfMissing(ctx, "country_dashboard_daily_metric", "lucky_gift_anchor_share",
"ALTER TABLE country_dashboard_daily_metric ADD COLUMN lucky_gift_anchor_share decimal(20,2) NOT NULL DEFAULT 0 AFTER lucky_gift_payout"); err != nil {
return err
}
if err := r.addColumnIfMissing(ctx, "country_dashboard_period_metric", "lucky_gift_anchor_share",
"ALTER TABLE country_dashboard_period_metric ADD COLUMN lucky_gift_anchor_share decimal(20,2) NOT NULL DEFAULT 0 AFTER lucky_gift_payout"); err != nil {
return err
}
return r.addIndexIfMissing(ctx, "dashboard_cdc_user_country", "idx_dashboard_cdc_user_country_origin_created", return r.addIndexIfMissing(ctx, "dashboard_cdc_user_country", "idx_dashboard_cdc_user_country_origin_created",
"ALTER TABLE dashboard_cdc_user_country ADD INDEX idx_dashboard_cdc_user_country_origin_created (sys_origin, user_created_at, country_code)") "ALTER TABLE dashboard_cdc_user_country ADD INDEX idx_dashboard_cdc_user_country_origin_created (sys_origin, user_created_at, country_code)")
} }