fix: reconcile dashboard recharge gaps

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)

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

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)
 	}

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
+}