From 905da26cf8644b14b726415c412b1f5a5e8fee62 Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 25 May 2026 11:16:45 +0800 Subject: [PATCH] fix: reconcile dashboard recharge gaps --- cmd/dashboard-cdc-worker/main.go | 2 + internal/cdc/handler.go | 16 ++ internal/config/config.go | 17 ++ internal/storage/recharge_reconcile.go | 314 +++++++++++++++++++++++++ 4 files changed, 349 insertions(+) create mode 100644 internal/storage/recharge_reconcile.go diff --git a/cmd/dashboard-cdc-worker/main.go b/cmd/dashboard-cdc-worker/main.go index e8914fa..6e565c4 100644 --- a/cmd/dashboard-cdc-worker/main.go +++ b/cmd/dashboard-cdc-worker/main.go @@ -80,6 +80,8 @@ func main() { } projector := dashboard.NewProjector(cfg, repo, redisClient, metrics) + repo.StartRechargeReconciler(ctx, logger, projector) + worker, err := cdc.NewWorker(cfg, repo, projector, logger, metrics) if err != nil { logger.Error("create cdc worker failed", "error", err) diff --git a/internal/cdc/handler.go b/internal/cdc/handler.go index d50e1a9..6cdad30 100644 --- a/internal/cdc/handler.go +++ b/internal/cdc/handler.go @@ -150,6 +150,9 @@ func (h *handler) applyRows(ctx context.Context, e *canal.RowsEvent, rowIndex in return nil } eventKey := fmt.Sprintf("%s:%d:%s:%s:%d", h.currentFile(), e.Header.LogPos, source, e.Action, rowIndex) + if key := stableRechargeEventKey(source, contributions); key != "" { + eventKey = key + } event := dashboard.Event{ EventKey: eventKey, Source: source, @@ -163,6 +166,19 @@ func (h *handler) applyRows(ctx context.Context, e *canal.RowsEvent, rowIndex in return nil } +func stableRechargeEventKey(source string, contributions []dashboard.Contribution) string { + if source != "likei.order_purchase_history" && source != "likei.order_user_purchase_pay" { + return "" + } + for _, contribution := range contributions { + if contribution.DeltaSign < 0 || contribution.SourcePK == "" || contribution.RechargeDetail == nil { + continue + } + return fmt.Sprintf("reconcile:%s:%s:complete", source, contribution.SourcePK) + } + return "" +} + func (h *handler) shouldFlush() bool { if len(h.batch) == 0 { return false diff --git a/internal/config/config.go b/internal/config/config.go index 26ceda5..77673b4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -15,6 +15,7 @@ type Config struct { Redis RedisConfig CDC CDCConfig Dashboard DashboardConfig + Recharge RechargeReconcileConfig Health HealthConfig } @@ -62,6 +63,13 @@ type DashboardConfig struct { CleanupBatchSize int } +type RechargeReconcileConfig struct { + Enabled bool + Interval time.Duration + CurrentDayLag time.Duration + BatchLimit int +} + type HealthConfig struct { Addr string } @@ -108,6 +116,12 @@ func Load() (Config, error) { UserMetricDays: envInt("DASHBOARD_USER_METRIC_RETENTION_DAYS", 120), CleanupBatchSize: envInt("DASHBOARD_CLEANUP_BATCH_SIZE", 5000), }, + Recharge: RechargeReconcileConfig{ + Enabled: envBool("DASHBOARD_RECHARGE_RECONCILE_ENABLED", true), + Interval: envDuration("DASHBOARD_RECHARGE_RECONCILE_INTERVAL", 10*time.Minute), + CurrentDayLag: envDuration("DASHBOARD_RECHARGE_RECONCILE_CURRENT_DAY_LAG", 15*time.Minute), + BatchLimit: envInt("DASHBOARD_RECHARGE_RECONCILE_BATCH_LIMIT", 5000), + }, Health: HealthConfig{ Addr: env("HEALTH_ADDR", ":2910"), }, @@ -134,6 +148,9 @@ func Load() (Config, error) { if cfg.Dashboard.CleanupBatchSize <= 0 { cfg.Dashboard.CleanupBatchSize = 5000 } + if cfg.Recharge.BatchLimit <= 0 { + cfg.Recharge.BatchLimit = 5000 + } if cfg.CDC.StartMode != "checkpoint" && cfg.CDC.StartMode != "current" { return Config{}, fmt.Errorf("unsupported CDC_START_MODE=%s", cfg.CDC.StartMode) } diff --git a/internal/storage/recharge_reconcile.go b/internal/storage/recharge_reconcile.go new file mode 100644 index 0000000..999f44e --- /dev/null +++ b/internal/storage/recharge_reconcile.go @@ -0,0 +1,314 @@ +package storage + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "strings" + "time" + + "github.com/shopspring/decimal" + + "dashboard-cdc-worker/internal/dashboard" +) + +const ( + reconcileOfficialSource = "likei.order_purchase_history" + reconcileMifapaySource = "likei.order_user_purchase_pay" +) + +type officialRechargeRow struct { + id string + userID int64 + sysOrigin string + payPlatform string + unitPrice decimal.Decimal + coinQuantity decimal.Decimal + eventTime time.Time + user dashboard.UserCountry +} + +type mifapayRechargeRow struct { + id string + userID int64 + sysOrigin string + amount decimal.Decimal + eventTime time.Time + user dashboard.UserCountry +} + +func (r *MySQLRepository) BuildRechargeReconcileEvents(ctx context.Context, since time.Time, + limit int) ([]dashboard.Event, error) { + if limit <= 0 { + limit = 5000 + } + officialRows, err := r.listOfficialRechargeRows(ctx, since, limit) + if err != nil { + return nil, err + } + remaining := limit - len(officialRows) + if remaining <= 0 { + remaining = 1 + } + mifapayRows, err := r.listMifapayRechargeRows(ctx, since, remaining) + if err != nil { + return nil, err + } + events := make([]dashboard.Event, 0, len(officialRows)+len(mifapayRows)) + for _, row := range officialRows { + if !r.allowedOrigin(row.sysOrigin) || row.user.IsDeleted || row.user.SysOrigin != row.sysOrigin { + continue + } + contribution := dashboard.Contribution{ + SourceSchema: "likei", + SourceTable: "order_purchase_history", + SourcePK: row.id, + SourceAction: "insert", + DeltaSign: 1, + SysOrigin: row.sysOrigin, + UserID: row.userID, + Country: row.user.Country, + EventTime: row.eventTime, + UserCreatedAt: row.user.CreatedAt, + NewUserRechargeCandidate: row.unitPrice, + RechargeDetail: officialRechargeDetail(row), + } + if row.payPlatform != "MIFA_PAY" { + contribution.OfficialRecharge = row.unitPrice + } + if row.payPlatform == "GOOGLE" { + contribution.GoogleRecharge = row.unitPrice + } + events = append(events, dashboard.Event{ + EventKey: rechargeReconcileEventKey(reconcileOfficialSource, row.id), + Source: reconcileOfficialSource, + Contributions: []dashboard.Contribution{contribution}, + }) + } + for _, row := range mifapayRows { + if !r.allowedOrigin(row.sysOrigin) || row.user.IsDeleted || row.user.SysOrigin != row.sysOrigin { + continue + } + contribution := dashboard.Contribution{ + SourceSchema: "likei", + SourceTable: "order_user_purchase_pay", + SourcePK: row.id, + SourceAction: "insert", + DeltaSign: 1, + SysOrigin: row.sysOrigin, + UserID: row.userID, + Country: row.user.Country, + EventTime: row.eventTime, + UserCreatedAt: row.user.CreatedAt, + NewUserRechargeCandidate: row.amount, + MifapayRecharge: row.amount, + RechargeDetail: mifapayRechargeDetail(row), + } + events = append(events, dashboard.Event{ + EventKey: rechargeReconcileEventKey(reconcileMifapaySource, row.id), + Source: reconcileMifapaySource, + Contributions: []dashboard.Contribution{contribution}, + }) + } + return events, nil +} + +func (r *MySQLRepository) StartRechargeReconciler(ctx context.Context, logger *slog.Logger, + projector interface { + ApplyEvents(context.Context, []dashboard.Event) error + }) { + if !r.cfg.Recharge.Enabled || r.cfg.Recharge.Interval <= 0 { + return + } + run := func() { + since := currentStatDayStart(r.cfg.Dashboard.StatTimezones, r.cfg.Recharge.CurrentDayLag) + events, err := r.BuildRechargeReconcileEvents(ctx, since, r.cfg.Recharge.BatchLimit) + if err != nil { + if r.metrics != nil { + r.metrics.RecordError(err) + } + logger.Error("dashboard recharge reconcile query failed", "error", err) + return + } + if len(events) == 0 { + return + } + if err := projector.ApplyEvents(ctx, events); err != nil { + if r.metrics != nil { + r.metrics.RecordError(err) + } + logger.Error("dashboard recharge reconcile apply failed", "events", len(events), "error", err) + return + } + logger.Info("dashboard recharge reconcile applied", "events", len(events), "since", since) + } + go func() { + run() + ticker := time.NewTicker(r.cfg.Recharge.Interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + run() + } + } + }() +} + +func (r *MySQLRepository) listOfficialRechargeRows(ctx context.Context, since time.Time, + limit int) ([]officialRechargeRow, error) { + rows, err := r.db.QueryContext(ctx, ` +SELECT CAST(o.id AS CHAR), o.user_id, o.sys_origin, o.pay_platform, o.unit_price, o.coin_quantity, o.create_time, + u.id, u.origin_sys, IFNULL(NULLIF(u.country_code, ''), ?), + IFNULL(NULLIF(u.country_name, ''), IFNULL(NULLIF(u.country_code, ''), ?)), + IFNULL(u.is_del, 0), u.create_time, u.update_time +FROM order_purchase_history o +JOIN user_base_info u ON u.id = o.user_id +WHERE o.create_time >= ? + AND o.evn = 'PROD' + AND o.status = 'COMPLETE' + AND IFNULL(o.is_trial_period, 0) = 0 +ORDER BY o.create_time ASC +LIMIT ?`, r.cfg.Dashboard.UnknownCountryCode, r.cfg.Dashboard.UnknownCountryCode, since, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var result []officialRechargeRow + for rows.Next() { + var item officialRechargeRow + if err := scanRechargeUser(rows, &item.id, &item.userID, &item.sysOrigin, + &item.payPlatform, &item.unitPrice, &item.coinQuantity, &item.eventTime, + &item.user); err != nil { + return nil, err + } + result = append(result, item) + } + return result, rows.Err() +} + +func (r *MySQLRepository) listMifapayRechargeRows(ctx context.Context, since time.Time, + limit int) ([]mifapayRechargeRow, error) { + rows, err := r.db.QueryContext(ctx, ` +SELECT CAST(o.id AS CHAR), o.user_id, o.sys_origin, o.compute_usd_amount, o.create_time, + u.id, u.origin_sys, IFNULL(NULLIF(u.country_code, ''), ?), + IFNULL(NULLIF(u.country_name, ''), IFNULL(NULLIF(u.country_code, ''), ?)), + IFNULL(u.is_del, 0), u.create_time, u.update_time +FROM order_user_purchase_pay o +JOIN user_base_info u ON u.id = o.user_id +WHERE o.create_time >= ? + AND o.evn = 'PROD' + AND o.factory_code = 'MIFA_PAY' + AND o.pay_status = 'SUCCESSFUL' + AND o.receipt_type = 'PAYMENT' + AND o.refund_status = 'NONE' +ORDER BY o.create_time ASC +LIMIT ?`, r.cfg.Dashboard.UnknownCountryCode, r.cfg.Dashboard.UnknownCountryCode, since, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var result []mifapayRechargeRow + for rows.Next() { + var item mifapayRechargeRow + var countryCode, countryName string + var isDeleted bool + var userCreatedAt, userUpdatedAt sql.NullTime + if err := rows.Scan(&item.id, &item.userID, &item.sysOrigin, &item.amount, + &item.eventTime, &item.user.UserID, &item.user.SysOrigin, &countryCode, + &countryName, &isDeleted, &userCreatedAt, &userUpdatedAt); err != nil { + return nil, err + } + item.user.Country = dashboard.Country{Code: countryCode, Name: countryName} + item.user.IsDeleted = isDeleted + if userCreatedAt.Valid { + item.user.CreatedAt = userCreatedAt.Time + } + if userUpdatedAt.Valid { + item.user.UpdatedAt = userUpdatedAt.Time + } + result = append(result, item) + } + return result, rows.Err() +} + +func scanRechargeUser(rows *sql.Rows, id *string, userID *int64, sysOrigin *string, + payPlatform *string, unitPrice *decimal.Decimal, coinQuantity *decimal.Decimal, + eventTime *time.Time, user *dashboard.UserCountry) error { + var countryCode, countryName string + var isDeleted bool + var userCreatedAt, userUpdatedAt sql.NullTime + if err := rows.Scan(id, userID, sysOrigin, payPlatform, unitPrice, coinQuantity, eventTime, + &user.UserID, &user.SysOrigin, &countryCode, &countryName, &isDeleted, + &userCreatedAt, &userUpdatedAt); err != nil { + return err + } + user.Country = dashboard.Country{Code: countryCode, Name: countryName} + user.IsDeleted = isDeleted + if userCreatedAt.Valid { + user.CreatedAt = userCreatedAt.Time + } + if userUpdatedAt.Valid { + user.UpdatedAt = userUpdatedAt.Time + } + return nil +} + +func officialRechargeDetail(row officialRechargeRow) *dashboard.RechargeDetail { + return &dashboard.RechargeDetail{ + SourceType: "OFFICIAL", + RecordID: row.id, + SourceChannel: row.payPlatform, + SysOrigin: row.user.SysOrigin, + UserID: row.userID, + Country: row.user.Country, + Amount: row.unitPrice, + CoinQuantity: row.coinQuantity, + EventTime: row.eventTime, + } +} + +func mifapayRechargeDetail(row mifapayRechargeRow) *dashboard.RechargeDetail { + return &dashboard.RechargeDetail{ + SourceType: "MIFAPAY", + RecordID: row.id, + SourceChannel: "MIFA_PAY", + SysOrigin: row.user.SysOrigin, + UserID: row.userID, + Country: row.user.Country, + Amount: row.amount, + EventTime: row.eventTime, + } +} + +func rechargeReconcileEventKey(source string, id string) string { + return fmt.Sprintf("reconcile:%s:%s:complete", source, id) +} + +func currentStatDayStart(statTimezones []string, lag time.Duration) time.Time { + timezone := "Asia/Riyadh" + if len(statTimezones) > 0 && strings.TrimSpace(statTimezones[0]) != "" { + timezone = strings.TrimSpace(statTimezones[0]) + } + location, err := time.LoadLocation(timezone) + if err != nil { + location = time.UTC + } + now := time.Now().In(location).Add(-lag) + return time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, location) +} + +func (r *MySQLRepository) allowedOrigin(sysOrigin string) bool { + if len(r.cfg.Dashboard.SysOrigins) == 0 { + return true + } + for _, origin := range r.cfg.Dashboard.SysOrigins { + if strings.TrimSpace(origin) == sysOrigin { + return true + } + } + return false +}