From db491b92e59608817a3a6c7a748571230a543a4c Mon Sep 17 00:00:00 2001 From: zhx Date: Sat, 23 May 2026 12:27:44 +0800 Subject: [PATCH] fix: stabilize dashboard cdc worker --- cmd/dashboard-cdc-worker/main.go | 11 +- env.example | 8 + internal/cdc/handler.go | 88 +++++- internal/cdc/worker.go | 44 ++- internal/config/config.go | 44 ++- internal/dashboard/mapper.go | 18 +- internal/dashboard/mapper_helpers.go | 27 ++ internal/dashboard/projector.go | 128 +++++--- internal/dashboard/store.go | 4 + internal/dashboard/types.go | 28 +- internal/health/server.go | 18 +- internal/monitor/metrics.go | 173 +++++++++++ internal/storage/batch.go | 380 ++++++++++++++++++++++++ internal/storage/batch_sql.go | 231 ++++++++++++++ internal/storage/cache.go | 58 ++++ internal/storage/cleanup.go | 119 ++++++++ internal/storage/game.go | 41 ++- internal/storage/recharge_detail.go | 45 +++ internal/storage/repository.go | 99 +++++- internal/storage/schema.go | 21 ++ internal/storage/user.go | 8 + migrations/001_dashboard_cdc_worker.sql | 20 ++ 22 files changed, 1527 insertions(+), 86 deletions(-) create mode 100644 internal/monitor/metrics.go create mode 100644 internal/storage/batch.go create mode 100644 internal/storage/batch_sql.go create mode 100644 internal/storage/cache.go create mode 100644 internal/storage/cleanup.go create mode 100644 internal/storage/recharge_detail.go diff --git a/cmd/dashboard-cdc-worker/main.go b/cmd/dashboard-cdc-worker/main.go index 344e2f6..e8914fa 100644 --- a/cmd/dashboard-cdc-worker/main.go +++ b/cmd/dashboard-cdc-worker/main.go @@ -17,6 +17,7 @@ import ( "dashboard-cdc-worker/internal/config" "dashboard-cdc-worker/internal/dashboard" "dashboard-cdc-worker/internal/health" + "dashboard-cdc-worker/internal/monitor" "dashboard-cdc-worker/internal/storage" ) @@ -42,7 +43,8 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() - healthServer := health.New(cfg.Health.Addr) + metrics := monitor.New() + healthServer := health.New(cfg.Health.Addr, metrics) healthServer.Start(logger) defer func() { shutdownCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) @@ -55,11 +57,12 @@ func main() { os.Exit(1) } - repo := storage.NewMySQLRepository(db, cfg) + repo := storage.NewMySQLRepository(db, cfg, metrics) if err := repo.EnsureSchema(ctx); err != nil { logger.Error("ensure schema failed", "error", err) os.Exit(1) } + repo.StartCleanup(ctx, logger) var redisClient *redis.Client if cfg.Redis.Enabled { @@ -76,8 +79,8 @@ func main() { } } - projector := dashboard.NewProjector(cfg, repo, redisClient) - worker, err := cdc.NewWorker(cfg, repo, projector, logger) + projector := dashboard.NewProjector(cfg, repo, redisClient, metrics) + worker, err := cdc.NewWorker(cfg, repo, projector, logger, metrics) if err != nil { logger.Error("create cdc worker failed", "error", err) os.Exit(1) diff --git a/env.example b/env.example index 2757a07..3d1e4fd 100644 --- a/env.example +++ b/env.example @@ -5,9 +5,17 @@ DASHBOARD_MYSQL_DSN=dashboard_cdc:password@tcp(127.0.0.1:3306)/likei?charset=utf CDC_START_MODE=checkpoint CDC_SERVER_ID=29301 CDC_DRY_RUN=true +CDC_BATCH_SIZE=1000 +CDC_FLUSH_INTERVAL=3s +CDC_BACKFILL_LOCK_WAIT=1s +CDC_MASTER_POSITION_POLL_INTERVAL=1m DASHBOARD_SYS_ORIGINS=LIKEI DASHBOARD_STAT_TIMEZONES=Asia/Riyadh DASHBOARD_STORAGE_TIMEZONE=Asia/Riyadh +DASHBOARD_CLEANUP_INTERVAL=30m +DASHBOARD_DEDUP_RETENTION=72h +DASHBOARD_USER_METRIC_RETENTION_DAYS=120 +DASHBOARD_CLEANUP_BATCH_SIZE=5000 REDIS_ENABLED=false REDIS_ADDR=127.0.0.1:6379 HEALTH_ADDR=:2910 diff --git a/internal/cdc/handler.go b/internal/cdc/handler.go index 284ad36..d50e1a9 100644 --- a/internal/cdc/handler.go +++ b/internal/cdc/handler.go @@ -5,6 +5,7 @@ import ( "fmt" "log/slog" "sync" + "time" "github.com/go-mysql-org/go-mysql/canal" "github.com/go-mysql-org/go-mysql/mysql" @@ -12,6 +13,7 @@ import ( "dashboard-cdc-worker/internal/config" "dashboard-cdc-worker/internal/dashboard" + "dashboard-cdc-worker/internal/monitor" ) type handler struct { @@ -23,15 +25,22 @@ type handler struct { logger *slog.Logger mu sync.RWMutex file string + batch []dashboard.Event + lastFlush time.Time + metrics *monitor.Metrics } -func newHandler(cfg config.Config, store dashboard.Store, projector *dashboard.Projector, logger *slog.Logger) *handler { +func newHandler(cfg config.Config, store dashboard.Store, projector *dashboard.Projector, + logger *slog.Logger, metrics *monitor.Metrics) *handler { return &handler{ cfg: cfg, store: store, projector: projector, mapper: dashboard.NewMapper(store, cfg.Dashboard.SysOrigins, cfg.Dashboard.UnknownCountryCode), logger: logger, + batch: make([]dashboard.Event, 0, cfg.CDC.BatchSize), + lastFlush: time.Now(), + metrics: metrics, } } @@ -47,18 +56,21 @@ func (h *handler) OnRow(e *canal.RowsEvent) error { case canal.InsertAction: for index, row := range e.Rows { if err := h.applyRows(ctx, e, index, nil, row); err != nil { + h.recordError(err) return err } } case canal.DeleteAction: for index, row := range e.Rows { if err := h.applyRows(ctx, e, index, row, nil); err != nil { + h.recordError(err) return err } } case canal.UpdateAction: for index := 0; index+1 < len(e.Rows); index += 2 { if err := h.applyRows(ctx, e, index/2, e.Rows[index], e.Rows[index+1]); err != nil { + h.recordError(err) return err } } @@ -72,10 +84,24 @@ func (h *handler) OnPosSynced(_ *replication.EventHeader, pos mysql.Position, _ if h.cfg.CDC.DryRun { return nil } - if !force && h.cfg.CDC.FlushInterval > 0 { - // canal already throttles calls; keep this hook simple and always persist when invoked. + if len(h.batch) > 0 { + if force || h.shouldFlush() { + if err := h.flushBatch(context.Background()); err != nil { + h.recordError(err) + return err + } + } else { + return nil + } } - return h.store.SaveCheckpoint(context.Background(), pos) + if err := h.store.SaveCheckpoint(context.Background(), pos); err != nil { + h.recordError(err) + return err + } + if h.metrics != nil { + h.metrics.SetLastSyncedPosition(pos) + } + return nil } func (h *handler) String() string { @@ -120,9 +146,55 @@ func (h *handler) applyRows(ctx context.Context, e *canal.RowsEvent, rowIndex in contributions = append(contributions, items...) } - eventKey := fmt.Sprintf("%s:%d:%s:%s:%d", h.currentFile(), e.Header.LogPos, source, e.Action, rowIndex) - if user != nil { - return h.projector.ApplyUser(ctx, eventKey, source, *user, contributions) + if user == nil && len(contributions) == 0 { + return nil + } + eventKey := fmt.Sprintf("%s:%d:%s:%s:%d", h.currentFile(), e.Header.LogPos, source, e.Action, rowIndex) + event := dashboard.Event{ + EventKey: eventKey, + Source: source, + User: user, + Contributions: contributions, + } + h.batch = append(h.batch, event) + if h.shouldFlush() { + return h.flushBatch(ctx) + } + return nil +} + +func (h *handler) shouldFlush() bool { + if len(h.batch) == 0 { + return false + } + if len(h.batch) >= h.cfg.CDC.BatchSize { + return true + } + return h.cfg.CDC.FlushInterval > 0 && time.Since(h.lastFlush) >= h.cfg.CDC.FlushInterval +} + +func (h *handler) flushBatch(ctx context.Context) error { + if len(h.batch) == 0 { + h.lastFlush = time.Now() + return nil + } + events := make([]dashboard.Event, len(h.batch)) + copy(events, h.batch) + start := time.Now() + if err := h.projector.ApplyEvents(ctx, events); err != nil { + return err + } + if h.metrics != nil { + h.metrics.RecordBatch(len(events), time.Since(start)) + } + h.logger.Info("dashboard cdc batch flushed", "events", len(events)) + h.batch = h.batch[:0] + h.lastFlush = time.Now() + return nil +} + +func (h *handler) recordError(err error) { + if h.metrics != nil { + h.metrics.RecordError(err) } - return h.projector.Apply(ctx, eventKey, source, contributions) } diff --git a/internal/cdc/worker.go b/internal/cdc/worker.go index 08e0d02..b42dc2c 100644 --- a/internal/cdc/worker.go +++ b/internal/cdc/worker.go @@ -12,6 +12,7 @@ import ( "dashboard-cdc-worker/internal/config" "dashboard-cdc-worker/internal/dashboard" + "dashboard-cdc-worker/internal/monitor" ) type Worker struct { @@ -20,9 +21,11 @@ type Worker struct { projector *dashboard.Projector logger *slog.Logger canal *canal.Canal + metrics *monitor.Metrics } -func NewWorker(cfg config.Config, store dashboard.Store, projector *dashboard.Projector, logger *slog.Logger) (*Worker, error) { +func NewWorker(cfg config.Config, store dashboard.Store, projector *dashboard.Projector, + logger *slog.Logger, metrics *monitor.Metrics) (*Worker, error) { canalCfg := canal.NewDefaultConfig() canalCfg.Addr = cfg.MySQL.CDCAddr canalCfg.User = cfg.MySQL.CDCUser @@ -51,16 +54,18 @@ func NewWorker(cfg config.Config, store dashboard.Store, projector *dashboard.Pr projector: projector, logger: logger, canal: instance, + metrics: metrics, }, nil } func (w *Worker) Run(ctx context.Context) error { - handler := newHandler(w.cfg, w.store, w.projector, w.logger) + handler := newHandler(w.cfg, w.store, w.projector, w.logger, w.metrics) w.canal.SetEventHandler(handler) go func() { <-ctx.Done() w.canal.Close() }() + w.startMasterPositionReporter(ctx) if w.cfg.CDC.StartMode == "checkpoint" { pos, ok, err := w.store.LoadCheckpoint(ctx) @@ -69,6 +74,9 @@ func (w *Worker) Run(ctx context.Context) error { } if ok && pos.Name != "" && pos.Pos > 0 { handler.setCurrentFile(pos.Name) + if w.metrics != nil { + w.metrics.SetStartPosition(pos) + } w.logger.Info("starting cdc from checkpoint", "position", pos.String()) return w.canal.RunFrom(pos) } @@ -79,10 +87,42 @@ func (w *Worker) Run(ctx context.Context) error { return fmt.Errorf("get master position: %w", err) } handler.setCurrentFile(pos.Name) + if w.metrics != nil { + w.metrics.SetStartPosition(pos) + w.metrics.SetMasterPosition(pos) + } + if w.cfg.CDC.StartMode == "current" { + if err := w.store.SaveCheckpoint(ctx, pos); err != nil { + return fmt.Errorf("save current checkpoint: %w", err) + } + } w.logger.Info("starting cdc from current master position", "position", pos.String()) return w.canal.RunFrom(pos) } +func (w *Worker) startMasterPositionReporter(ctx context.Context) { + if w.metrics == nil || w.cfg.CDC.MasterPollEvery <= 0 { + return + } + go func() { + ticker := time.NewTicker(w.cfg.CDC.MasterPollEvery) + defer ticker.Stop() + for { + pos, err := w.canal.GetMasterPos() + if err == nil { + w.metrics.SetMasterPosition(pos) + } else { + w.metrics.RecordError(err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() +} + func includeRegex(tables []string) []string { out := make([]string, 0, len(tables)) for _, table := range tables { diff --git a/internal/config/config.go b/internal/config/config.go index c371c84..26ceda5 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -38,12 +38,15 @@ type RedisConfig struct { } type CDCConfig struct { - ServerID uint32 - StartMode string - DryRun bool - IncludeTables []string - FlushInterval time.Duration - HeartbeatEvery time.Duration + ServerID uint32 + StartMode string + DryRun bool + IncludeTables []string + FlushInterval time.Duration + HeartbeatEvery time.Duration + BatchSize int + BackfillLockWait time.Duration + MasterPollEvery time.Duration } type DashboardConfig struct { @@ -53,6 +56,10 @@ type DashboardConfig struct { RedisKeyPrefix string RealtimeRedisTTL time.Duration UnknownCountryCode string + CleanupInterval time.Duration + DedupRetention time.Duration + UserMetricDays int + CleanupBatchSize int } type HealthConfig struct { @@ -79,12 +86,15 @@ func Load() (Config, error) { DB: envInt("REDIS_DB", 0), }, CDC: CDCConfig{ - ServerID: uint32(envInt("CDC_SERVER_ID", 29301)), - StartMode: strings.ToLower(env("CDC_START_MODE", "checkpoint")), - DryRun: envBool("CDC_DRY_RUN", true), - IncludeTables: csv(env("CDC_INCLUDE_TABLES", defaultTables())), - FlushInterval: envDuration("CDC_FLUSH_INTERVAL", 3*time.Second), - HeartbeatEvery: envDuration("CDC_HEARTBEAT_EVERY", 30*time.Second), + ServerID: uint32(envInt("CDC_SERVER_ID", 29301)), + StartMode: strings.ToLower(env("CDC_START_MODE", "checkpoint")), + DryRun: envBool("CDC_DRY_RUN", true), + IncludeTables: csv(env("CDC_INCLUDE_TABLES", defaultTables())), + FlushInterval: envDuration("CDC_FLUSH_INTERVAL", 3*time.Second), + HeartbeatEvery: envDuration("CDC_HEARTBEAT_EVERY", 30*time.Second), + BatchSize: envInt("CDC_BATCH_SIZE", 1000), + BackfillLockWait: envDuration("CDC_BACKFILL_LOCK_WAIT", time.Second), + MasterPollEvery: envDuration("CDC_MASTER_POSITION_POLL_INTERVAL", time.Minute), }, Dashboard: DashboardConfig{ SysOrigins: csv(env("DASHBOARD_SYS_ORIGINS", "LIKEI")), @@ -93,6 +103,10 @@ func Load() (Config, error) { RedisKeyPrefix: env("DASHBOARD_REDIS_KEY_PREFIX", "country_dashboard"), RealtimeRedisTTL: envDuration("DASHBOARD_REALTIME_REDIS_TTL", 48*time.Hour), UnknownCountryCode: env("DASHBOARD_UNKNOWN_COUNTRY_CODE", "UNKNOWN"), + CleanupInterval: envDuration("DASHBOARD_CLEANUP_INTERVAL", 30*time.Minute), + DedupRetention: envDuration("DASHBOARD_DEDUP_RETENTION", 72*time.Hour), + UserMetricDays: envInt("DASHBOARD_USER_METRIC_RETENTION_DAYS", 120), + CleanupBatchSize: envInt("DASHBOARD_CLEANUP_BATCH_SIZE", 5000), }, Health: HealthConfig{ Addr: env("HEALTH_ADDR", ":2910"), @@ -114,6 +128,12 @@ func Load() (Config, error) { if cfg.CDC.ServerID == 0 { return Config{}, errors.New("CDC_SERVER_ID must be greater than 0") } + if cfg.CDC.BatchSize <= 0 { + cfg.CDC.BatchSize = 1000 + } + if cfg.Dashboard.CleanupBatchSize <= 0 { + cfg.Dashboard.CleanupBatchSize = 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/dashboard/mapper.go b/internal/dashboard/mapper.go index 74e9035..a92663c 100644 --- a/internal/dashboard/mapper.go +++ b/internal/dashboard/mapper.go @@ -89,6 +89,8 @@ func (m *Mapper) mapOfficialRecharge(ctx context.Context, schema string, table s c.GoogleRecharge = amount } c.NewUserRechargeCandidate = amount + c.RechargeDetail = rechargeDetail("OFFICIAL", rowPK(row), payPlatform, row, user, amount, + row.Decimal("coin_quantity"), "") return []Contribution{c}, nil } @@ -98,18 +100,30 @@ func (m *Mapper) mapMifapayRecharge(ctx context.Context, schema string, table st row.String("refund_status") != "NONE" { return nil, nil } - return m.amountContribution(ctx, schema, table, action, row, row.Decimal("compute_usd_amount"), func(c *Contribution, amount decimal.Decimal) { + items, err := m.amountContribution(ctx, schema, table, action, row, row.Decimal("compute_usd_amount"), func(c *Contribution, amount decimal.Decimal) { c.MifapayRecharge = amount }) + if err != nil || len(items) == 0 { + return items, err + } + items[0].RechargeDetail = rechargeDetail("MIFAPAY", rowPK(row), row.String("factory_code"), row, + detailUser(items[0]), row.Decimal("compute_usd_amount"), decimal.Zero, "") + return items, nil } func (m *Mapper) mapDealerRecharge(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { if row.String("origin") != "PURCHASE" || row.Int64("type") != 0 || !row.Decimal("amount").GreaterThan(decimal.Zero) { return nil, nil } - return m.amountContribution(ctx, schema, table, action, row, row.Decimal("amount"), func(c *Contribution, amount decimal.Decimal) { + items, err := m.amountContribution(ctx, schema, table, action, row, row.Decimal("amount"), func(c *Contribution, amount decimal.Decimal) { c.DealerRecharge = amount }) + if err != nil || len(items) == 0 { + return items, err + } + items[0].RechargeDetail = rechargeDetail("DEALER", rowPK(row), row.String("recharge_type"), row, + detailUser(items[0]), row.Decimal("amount"), row.Decimal("quantity"), row.String("remark")) + return items, nil } func (m *Mapper) mapSalaryExchange(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { diff --git a/internal/dashboard/mapper_helpers.go b/internal/dashboard/mapper_helpers.go index 617b0d1..316df0f 100644 --- a/internal/dashboard/mapper_helpers.go +++ b/internal/dashboard/mapper_helpers.go @@ -90,6 +90,33 @@ func rowPK(row Row) string { return strings.TrimSpace(row.String("event_id")) } +func rechargeDetail(sourceType string, recordID string, sourceChannel string, row Row, user UserCountry, + amount decimal.Decimal, coinQuantity decimal.Decimal, remark string) *RechargeDetail { + if recordID == "" || user.UserID == 0 { + return nil + } + return &RechargeDetail{ + SourceType: sourceType, + RecordID: recordID, + SourceChannel: strings.TrimSpace(sourceChannel), + SysOrigin: user.SysOrigin, + UserID: user.UserID, + Country: user.Country, + Amount: amount, + CoinQuantity: coinQuantity, + Remark: strings.TrimSpace(remark), + EventTime: row.Time("create_time"), + } +} + +func detailUser(c Contribution) UserCountry { + return UserCountry{ + UserID: c.UserID, + SysOrigin: c.SysOrigin, + Country: c.Country, + } +} + func signForAction(action string) int { if action == "delete" { return -1 diff --git a/internal/dashboard/projector.go b/internal/dashboard/projector.go index e5de125..bedf224 100644 --- a/internal/dashboard/projector.go +++ b/internal/dashboard/projector.go @@ -3,12 +3,14 @@ package dashboard import ( "context" "database/sql" + "errors" "log/slog" "time" "github.com/redis/go-redis/v9" "dashboard-cdc-worker/internal/config" + "dashboard-cdc-worker/internal/monitor" ) type Projector struct { @@ -16,21 +18,17 @@ type Projector struct { store Store redisClient *redis.Client storageZone *time.Location - statZones map[string]*time.Location + metrics *monitor.Metrics } -func NewProjector(cfg config.Config, store Store, redisClient *redis.Client) *Projector { +func NewProjector(cfg config.Config, store Store, redisClient *redis.Client, metrics *monitor.Metrics) *Projector { storageZone := mustLocation(cfg.Dashboard.StorageTimezone) - statZones := make(map[string]*time.Location, len(cfg.Dashboard.StatTimezones)) - for _, zone := range cfg.Dashboard.StatTimezones { - statZones[zone] = mustLocation(zone) - } return &Projector{ cfg: cfg, store: store, redisClient: redisClient, storageZone: storageZone, - statZones: statZones, + metrics: metrics, } } @@ -43,53 +41,93 @@ func (p *Projector) ApplyUser(ctx context.Context, eventKey string, source strin } func (p *Projector) apply(ctx context.Context, eventKey string, source string, user *UserCountry, contributions []Contribution) error { - if user == nil && len(contributions) == 0 { + event := Event{EventKey: eventKey, Source: source, User: user, Contributions: contributions} + return p.ApplyEvents(ctx, []Event{event}) +} + +func (p *Projector) ApplyEvents(ctx context.Context, events []Event) error { + if len(events) == 0 { return nil } - if len(contributions) == 0 { - contributions = nil + events = compactEvents(events) + if len(events) == 0 { + return nil } if p.cfg.CDC.DryRun { - slog.Info("dashboard cdc dry run", "eventKey", eventKey, "source", source, "user", user != nil, "contributions", len(contributions)) + slog.Info("dashboard cdc dry run batch", "events", len(events)) return nil } - if err := p.store.WithTx(ctx, func(tx *sql.Tx) error { - applied, err := p.store.MarkEventApplied(ctx, tx, eventKey, source) - if err != nil { - return err - } - if !applied { + + var appliedContributions []Contribution + err := p.withDashboardLock(ctx, func() error { + return p.store.WithTx(ctx, func(tx *sql.Tx) error { + batchContributions := make([]Contribution, 0, len(events)) + for _, event := range events { + applied, err := p.store.MarkEventApplied(ctx, tx, event.EventKey, event.Source) + if err != nil { + return err + } + if !applied { + continue + } + if event.User != nil { + if err := p.store.UpsertUserCountry(ctx, tx, *event.User); err != nil { + return err + } + } + for _, contribution := range event.Contributions { + if contribution.Empty() { + continue + } + if contribution.DeltaSign == 0 { + contribution.DeltaSign = 1 + } + batchContributions = append(batchContributions, contribution) + } + } + if err := p.store.ApplyContributionBatch(ctx, tx, batchContributions, + p.cfg.Dashboard.StorageTimezone, p.cfg.Dashboard.StatTimezones); err != nil { + return err + } + appliedContributions = batchContributions return nil - } - if user != nil { - if err := p.store.UpsertUserCountry(ctx, tx, *user); err != nil { - return err - } - } - for _, contribution := range contributions { - if contribution.Empty() { - continue - } - if contribution.DeltaSign == 0 { - contribution.DeltaSign = 1 - } - if err := p.store.ApplyDailyContribution(ctx, tx, contribution, p.cfg.Dashboard.StorageTimezone); err != nil { - return err - } - for statTimezone := range p.statZones { - if err := p.store.ApplyPeriodContribution(ctx, tx, contribution, statTimezone); err != nil { - return err - } - if err := p.store.ApplyGamePeriodContribution(ctx, tx, contribution, statTimezone); err != nil { - return err - } - } - } - return nil - }); err != nil { + }) + }) + if err != nil { return err } - return p.writeRealtimeRedis(ctx, contributions) + return p.writeRealtimeRedis(ctx, appliedContributions) +} + +func (p *Projector) withDashboardLock(ctx context.Context, fn func() error) error { + for { + err := p.store.WithDashboardLock(ctx, p.cfg.CDC.BackfillLockWait, fn) + if !errors.Is(err, ErrDashboardBackfillLocked) { + return err + } + if p.metrics != nil { + p.metrics.RecordLockWait() + } + slog.Warn("dashboard cdc paused because backfill lock is held") + timer := time.NewTimer(p.cfg.CDC.FlushInterval) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-timer.C: + } + } +} + +func compactEvents(events []Event) []Event { + out := make([]Event, 0, len(events)) + for _, event := range events { + if event.User == nil && len(event.Contributions) == 0 { + continue + } + out = append(out, event) + } + return out } func mustLocation(name string) *time.Location { diff --git a/internal/dashboard/store.go b/internal/dashboard/store.go index 48a791c..d9eb8c9 100644 --- a/internal/dashboard/store.go +++ b/internal/dashboard/store.go @@ -3,15 +3,18 @@ package dashboard import ( "context" "database/sql" + "time" "github.com/go-mysql-org/go-mysql/mysql" ) type Store interface { WithTx(ctx context.Context, fn func(*sql.Tx) error) error + WithDashboardLock(ctx context.Context, timeout time.Duration, fn func() error) error MarkEventApplied(ctx context.Context, tx *sql.Tx, eventKey string, source string) (bool, error) SaveCheckpoint(ctx context.Context, pos mysql.Position) error LoadCheckpoint(ctx context.Context) (mysql.Position, bool, error) + Cleanup(ctx context.Context) (int64, error) UpsertUserCountry(ctx context.Context, tx *sql.Tx, user UserCountry) error UserCountry(ctx context.Context, userID int64) (UserCountry, bool, error) @@ -20,4 +23,5 @@ type Store interface { ApplyDailyContribution(ctx context.Context, tx *sql.Tx, c Contribution, storageZone string) error ApplyPeriodContribution(ctx context.Context, tx *sql.Tx, c Contribution, statTimezone string) error ApplyGamePeriodContribution(ctx context.Context, tx *sql.Tx, c Contribution, statTimezone string) error + ApplyContributionBatch(ctx context.Context, tx *sql.Tx, contributions []Contribution, storageTimezone string, statTimezones []string) error } diff --git a/internal/dashboard/types.go b/internal/dashboard/types.go index 8e7cb15..87d0240 100644 --- a/internal/dashboard/types.go +++ b/internal/dashboard/types.go @@ -1,11 +1,14 @@ package dashboard import ( + "errors" "time" "github.com/shopspring/decimal" ) +var ErrDashboardBackfillLocked = errors.New("dashboard backfill lock is held") + const ( PeriodDay = "DAY" PeriodWeek = "WEEK" @@ -58,6 +61,15 @@ type Contribution struct { GamePayoutBy decimal.Decimal GameProfit decimal.Decimal GameOrder int64 + + RechargeDetail *RechargeDetail +} + +type Event struct { + EventKey string + Source string + User *UserCountry + Contributions []Contribution } func (c Contribution) Empty() bool { @@ -78,7 +90,21 @@ func (c Contribution) Empty() bool { c.GameConsume.IsZero() && c.GamePayoutBy.IsZero() && c.GameProfit.IsZero() && - c.GameOrder == 0 + c.GameOrder == 0 && + c.RechargeDetail == nil +} + +type RechargeDetail struct { + SourceType string + RecordID string + SourceChannel string + SysOrigin string + UserID int64 + Country Country + Amount decimal.Decimal + CoinQuantity decimal.Decimal + Remark string + EventTime time.Time } type PeriodWindow struct { diff --git a/internal/health/server.go b/internal/health/server.go index c6da19a..7328b31 100644 --- a/internal/health/server.go +++ b/internal/health/server.go @@ -7,17 +7,31 @@ import ( "log/slog" "net/http" "time" + + "dashboard-cdc-worker/internal/monitor" ) type Server struct { server *http.Server } -func New(addr string) *Server { +func New(addr string, metrics *monitor.Metrics) *Server { mux := http.NewServeMux() mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(map[string]string{"code": "ok"}) + payload := map[string]interface{}{"code": "ok"} + if metrics != nil { + payload["metrics"] = metrics.Snapshot() + } + _ = json.NewEncoder(w).Encode(payload) + }) + mux.HandleFunc("/metrics", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + if metrics == nil { + _ = json.NewEncoder(w).Encode(map[string]string{"code": "ok"}) + return + } + _ = json.NewEncoder(w).Encode(metrics.Snapshot()) }) return &Server{server: &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second}} } diff --git a/internal/monitor/metrics.go b/internal/monitor/metrics.go new file mode 100644 index 0000000..355fe42 --- /dev/null +++ b/internal/monitor/metrics.go @@ -0,0 +1,173 @@ +package monitor + +import ( + "strconv" + "strings" + "sync" + "time" + + "github.com/go-mysql-org/go-mysql/mysql" +) + +type Metrics struct { + mu sync.RWMutex + + startedAt time.Time + + masterPosition mysql.Position + checkpointPosition mysql.Position + lastSyncedPosition mysql.Position + startPosition mysql.Position + + lastFlushAt time.Time + lastFlushEvents int + totalFlushes uint64 + totalEvents uint64 + totalRetries uint64 + totalLockWaits uint64 + totalCleanupRows uint64 + lastCleanupAt time.Time + lastCleanupRows int64 + lastError string + lastErrorAt time.Time + lastBatchDuration time.Duration +} + +func New() *Metrics { + return &Metrics{startedAt: time.Now()} +} + +func (m *Metrics) SetStartPosition(pos mysql.Position) { + m.mu.Lock() + defer m.mu.Unlock() + m.startPosition = pos +} + +func (m *Metrics) SetMasterPosition(pos mysql.Position) { + m.mu.Lock() + defer m.mu.Unlock() + m.masterPosition = pos +} + +func (m *Metrics) SetCheckpointPosition(pos mysql.Position) { + m.mu.Lock() + defer m.mu.Unlock() + m.checkpointPosition = pos +} + +func (m *Metrics) SetLastSyncedPosition(pos mysql.Position) { + m.mu.Lock() + defer m.mu.Unlock() + m.lastSyncedPosition = pos +} + +func (m *Metrics) RecordBatch(events int, duration time.Duration) { + m.mu.Lock() + defer m.mu.Unlock() + m.lastFlushAt = time.Now() + m.lastFlushEvents = events + m.totalFlushes++ + m.totalEvents += uint64(events) + m.lastBatchDuration = duration +} + +func (m *Metrics) RecordRetry() { + m.mu.Lock() + defer m.mu.Unlock() + m.totalRetries++ +} + +func (m *Metrics) RecordLockWait() { + m.mu.Lock() + defer m.mu.Unlock() + m.totalLockWaits++ +} + +func (m *Metrics) RecordCleanup(rows int64) { + m.mu.Lock() + defer m.mu.Unlock() + m.lastCleanupAt = time.Now() + m.lastCleanupRows = rows + if rows > 0 { + m.totalCleanupRows += uint64(rows) + } +} + +func (m *Metrics) RecordError(err error) { + if err == nil { + return + } + m.mu.Lock() + defer m.mu.Unlock() + m.lastError = err.Error() + m.lastErrorAt = time.Now() +} + +func (m *Metrics) Snapshot() map[string]interface{} { + m.mu.RLock() + defer m.mu.RUnlock() + + return map[string]interface{}{ + "started_at": formatTime(m.startedAt), + "start_position": positionMap(m.startPosition), + "master_position": positionMap(m.masterPosition), + "checkpoint_position": positionMap(m.checkpointPosition), + "last_synced_position": positionMap(m.lastSyncedPosition), + "binlog_file_gap": fileGap(m.masterPosition, m.checkpointPosition), + "binlog_position_gap": positionGap(m.masterPosition, m.checkpointPosition), + "last_flush_at": formatTime(m.lastFlushAt), + "last_flush_events": m.lastFlushEvents, + "last_batch_duration_ms": m.lastBatchDuration.Milliseconds(), + "total_flushes": m.totalFlushes, + "total_events": m.totalEvents, + "total_retries": m.totalRetries, + "total_lock_waits": m.totalLockWaits, + "last_cleanup_at": formatTime(m.lastCleanupAt), + "last_cleanup_rows": m.lastCleanupRows, + "total_cleanup_rows": m.totalCleanupRows, + "last_error": m.lastError, + "last_error_at": formatTime(m.lastErrorAt), + } +} + +func positionMap(pos mysql.Position) map[string]interface{} { + return map[string]interface{}{ + "file": pos.Name, + "pos": pos.Pos, + } +} + +func positionGap(master mysql.Position, checkpoint mysql.Position) interface{} { + if master.Name == "" || checkpoint.Name == "" || master.Name != checkpoint.Name || master.Pos < checkpoint.Pos { + return nil + } + return master.Pos - checkpoint.Pos +} + +func fileGap(master mysql.Position, checkpoint mysql.Position) interface{} { + masterSeq, ok := binlogSequence(master.Name) + if !ok { + return nil + } + checkpointSeq, ok := binlogSequence(checkpoint.Name) + if !ok { + return nil + } + return masterSeq - checkpointSeq +} + +func binlogSequence(file string) (int64, bool) { + index := strings.LastIndex(file, ".") + if index < 0 || index == len(file)-1 { + return 0, false + } + value, err := strconv.ParseInt(file[index+1:], 10, 64) + return value, err == nil +} + +func formatTime(value time.Time) string { + if value.IsZero() { + return "" + } + return value.Format(time.RFC3339) +} diff --git a/internal/storage/batch.go b/internal/storage/batch.go new file mode 100644 index 0000000..1590849 --- /dev/null +++ b/internal/storage/batch.go @@ -0,0 +1,380 @@ +package storage + +import ( + "context" + "database/sql" + "strings" + "time" + + "github.com/shopspring/decimal" + + "dashboard-cdc-worker/internal/dashboard" +) + +const bulkInsertChunkSize = 500 + +type amountTotals struct { + newUserRecharge decimal.Decimal + officialRecharge decimal.Decimal + mifapayRecharge decimal.Decimal + googleRecharge decimal.Decimal + dealerRecharge decimal.Decimal + salaryExchange decimal.Decimal + luckyGiftTotalFlow decimal.Decimal + luckyGiftPayout decimal.Decimal + gameTotalFlow decimal.Decimal + gamePayout decimal.Decimal +} + +type gameAmountTotals struct { + consume decimal.Decimal + payout decimal.Decimal + profit decimal.Decimal + orders int64 +} + +type dailyAmountKey struct { + day time.Time + sysOrigin string + countryCode string + countryName string +} + +type periodAmountKey struct { + periodType string + periodKey string + statTimezone string + periodName string + startDate time.Time + endDate time.Time + sysOrigin string + countryCode string + countryName string +} + +type gameAmountKey struct { + periodType string + periodKey string + statTimezone string + periodName string + startDate time.Time + endDate time.Time + sysOrigin string + countryCode string + countryName string + gameProvider string + gameID string + gameName string +} + +type dailyUserKey struct { + day time.Time + sysOrigin string + countryCode string + countryName string + metricType string +} + +type periodUserKey struct { + periodType string + periodKey string + statTimezone string + periodName string + startDate time.Time + endDate time.Time + sysOrigin string + countryCode string + countryName string + metricType string +} + +type gameUserKey struct { + periodType string + periodKey string + statTimezone string + periodName string + startDate time.Time + endDate time.Time + sysOrigin string + countryCode string + countryName string + gameProvider string + gameID string + gameName string +} + +func (r *MySQLRepository) ApplyContributionBatch(ctx context.Context, tx *sql.Tx, + contributions []dashboard.Contribution, storageTimezone string, statTimezones []string) error { + if len(contributions) == 0 { + return nil + } + storageZone, err := time.LoadLocation(storageTimezone) + if err != nil { + storageZone = time.UTC + } + statZones := make(map[string]*time.Location, len(statTimezones)) + for _, statTimezone := range statTimezones { + zone, err := time.LoadLocation(statTimezone) + if err != nil { + zone = time.UTC + } + statZones[statTimezone] = zone + } + + dailyAmounts := make(map[dailyAmountKey]amountTotals) + periodAmounts := make(map[periodAmountKey]amountTotals) + gameAmounts := make(map[gameAmountKey]gameAmountTotals) + dailyUsers := make(map[dailyUserKey]map[int64]struct{}) + periodUsers := make(map[periodUserKey]map[int64]struct{}) + gameUsers := make(map[gameUserKey]map[int64]struct{}) + var negative []dashboard.Contribution + + for _, c := range contributions { + if err := r.applyRechargeDetail(ctx, tx, c); err != nil { + return err + } + if c.EventTime.IsZero() { + continue + } + if c.DeltaSign < 0 { + negative = append(negative, c) + continue + } + if c.DeltaSign == 0 { + c.DeltaSign = 1 + } + day := dashboard.DailyDate(c.EventTime, storageZone) + newUserRecharge := decimal.Zero + if dashboard.SamePeriod(c.UserCreatedAt, c.EventTime, dashboard.PeriodWindow{Type: dashboard.PeriodDay}, storageZone) { + newUserRecharge = c.NewUserRechargeCandidate + } + dailyKey := dailyAmountKey{day: day, sysOrigin: c.SysOrigin, countryCode: c.Country.Code, countryName: c.Country.Name} + totals := dailyAmounts[dailyKey] + totals.add(c, newUserRecharge) + dailyAmounts[dailyKey] = totals + addDailyUserGroups(dailyUsers, c, day) + + for statTimezone, zone := range statZones { + for _, window := range dashboard.PeriodWindows(c.EventTime, zone) { + startDate, endDate := windowDates(window) + newUserRecharge = decimal.Zero + if dashboard.SamePeriod(c.UserCreatedAt, c.EventTime, window, zone) { + newUserRecharge = c.NewUserRechargeCandidate + } + periodKey := periodAmountKey{ + periodType: window.Type, periodKey: window.Key, statTimezone: statTimezone, + periodName: window.Name, startDate: startDate, endDate: endDate, + sysOrigin: c.SysOrigin, countryCode: c.Country.Code, countryName: c.Country.Name, + } + periodTotal := periodAmounts[periodKey] + periodTotal.add(c, newUserRecharge) + periodAmounts[periodKey] = periodTotal + addPeriodUserGroups(periodUsers, c, window, statTimezone) + + if c.GameProvider != "" && c.GameID != "" { + gameKey := gameAmountKey{ + periodType: window.Type, periodKey: window.Key, statTimezone: statTimezone, + periodName: window.Name, startDate: startDate, endDate: endDate, + sysOrigin: c.SysOrigin, countryCode: c.Country.Code, countryName: c.Country.Name, + gameProvider: c.GameProvider, gameID: c.GameID, gameName: c.GameName, + } + gameTotal := gameAmounts[gameKey] + gameTotal.add(c) + gameAmounts[gameKey] = gameTotal + if c.GameUser { + addGameUserGroup(gameUsers, c, window, statTimezone) + } + } + } + } + } + + for key, totals := range dailyAmounts { + if totals.empty() { + continue + } + if err := r.upsertDailyAmountTotals(ctx, tx, key, totals); err != nil { + return err + } + } + for key, totals := range periodAmounts { + if totals.empty() { + continue + } + if err := r.upsertPeriodAmountTotals(ctx, tx, key, totals); err != nil { + return err + } + } + for key, totals := range gameAmounts { + if totals.empty() { + continue + } + if err := r.upsertGameAmountTotals(ctx, tx, key, totals); err != nil { + return err + } + } + for key, users := range dailyUsers { + if err := r.applyDailyUserGroup(ctx, tx, key, users); err != nil { + return err + } + } + for key, users := range periodUsers { + if err := r.applyPeriodUserGroup(ctx, tx, key, users); err != nil { + return err + } + } + for key, users := range gameUsers { + if err := r.applyGameUserGroup(ctx, tx, key, users); err != nil { + return err + } + } + for _, c := range negative { + if err := r.ApplyDailyContribution(ctx, tx, c, storageTimezone); err != nil { + return err + } + for _, statTimezone := range statTimezones { + if err := r.ApplyPeriodContribution(ctx, tx, c, statTimezone); err != nil { + return err + } + if err := r.ApplyGamePeriodContribution(ctx, tx, c, statTimezone); err != nil { + return err + } + } + } + return nil +} + +func (t *amountTotals) add(c dashboard.Contribution, newUserRecharge decimal.Decimal) { + sign := decimal.NewFromInt(int64(c.DeltaSign)) + t.newUserRecharge = t.newUserRecharge.Add(newUserRecharge.Mul(sign)) + t.officialRecharge = t.officialRecharge.Add(c.OfficialRecharge.Mul(sign)) + t.mifapayRecharge = t.mifapayRecharge.Add(c.MifapayRecharge.Mul(sign)) + t.googleRecharge = t.googleRecharge.Add(c.GoogleRecharge.Mul(sign)) + t.dealerRecharge = t.dealerRecharge.Add(c.DealerRecharge.Mul(sign)) + t.salaryExchange = t.salaryExchange.Add(c.SalaryExchange.Mul(sign)) + t.luckyGiftTotalFlow = t.luckyGiftTotalFlow.Add(c.LuckyGiftTotalFlow.Mul(sign)) + t.luckyGiftPayout = t.luckyGiftPayout.Add(c.LuckyGiftPayout.Mul(sign)) + t.gameTotalFlow = t.gameTotalFlow.Add(c.GameTotalFlow.Mul(sign)) + t.gamePayout = t.gamePayout.Add(c.GamePayout.Mul(sign)) +} + +func (t amountTotals) empty() bool { + return t.newUserRecharge.IsZero() && t.officialRecharge.IsZero() && t.mifapayRecharge.IsZero() && + t.googleRecharge.IsZero() && t.dealerRecharge.IsZero() && t.salaryExchange.IsZero() && + t.luckyGiftTotalFlow.IsZero() && t.luckyGiftPayout.IsZero() && + t.gameTotalFlow.IsZero() && t.gamePayout.IsZero() +} + +func (t *gameAmountTotals) add(c dashboard.Contribution) { + sign := decimal.NewFromInt(int64(c.DeltaSign)) + t.consume = t.consume.Add(c.GameConsume.Mul(sign)) + t.payout = t.payout.Add(c.GamePayoutBy.Mul(sign)) + t.profit = t.profit.Add(c.GameProfit.Mul(sign)) + t.orders += c.GameOrder * int64(c.DeltaSign) +} + +func (t gameAmountTotals) empty() bool { + return t.consume.IsZero() && t.payout.IsZero() && t.profit.IsZero() && t.orders == 0 +} + +func addDailyUserGroups(groups map[dailyUserKey]map[int64]struct{}, c dashboard.Contribution, day time.Time) { + if c.UserID == 0 { + return + } + if c.CountryNewUser > 0 { + addDailyUser(groups, c, day, dashboard.MetricCountryNewUser) + } + if c.LuckyGiftUser { + addDailyUser(groups, c, day, dashboard.MetricLuckyGiftUser) + } + if c.GameUser { + addDailyUser(groups, c, day, dashboard.MetricGameUser) + } +} + +func addPeriodUserGroups(groups map[periodUserKey]map[int64]struct{}, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string) { + if c.UserID == 0 { + return + } + if c.CountryNewUser > 0 { + addPeriodUser(groups, c, window, statTimezone, dashboard.MetricCountryNewUser) + } + if c.LuckyGiftUser { + addPeriodUser(groups, c, window, statTimezone, dashboard.MetricLuckyGiftUser) + } + if c.GameUser { + addPeriodUser(groups, c, window, statTimezone, dashboard.MetricGameUser) + } +} + +func addDailyUser(groups map[dailyUserKey]map[int64]struct{}, c dashboard.Contribution, day time.Time, metricType string) { + key := dailyUserKey{day: day, sysOrigin: c.SysOrigin, countryCode: c.Country.Code, countryName: c.Country.Name, metricType: metricType} + addUserID(groups, key, c.UserID) +} + +func addPeriodUser(groups map[periodUserKey]map[int64]struct{}, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string, metricType string) { + startDate, endDate := windowDates(window) + key := periodUserKey{ + periodType: window.Type, periodKey: window.Key, statTimezone: statTimezone, periodName: window.Name, + startDate: startDate, endDate: endDate, sysOrigin: c.SysOrigin, countryCode: c.Country.Code, + countryName: c.Country.Name, metricType: metricType, + } + addUserID(groups, key, c.UserID) +} + +func addGameUserGroup(groups map[gameUserKey]map[int64]struct{}, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string) { + if c.UserID == 0 { + return + } + startDate, endDate := windowDates(window) + key := gameUserKey{ + periodType: window.Type, periodKey: window.Key, statTimezone: statTimezone, periodName: window.Name, + startDate: startDate, endDate: endDate, sysOrigin: c.SysOrigin, countryCode: c.Country.Code, + countryName: c.Country.Name, gameProvider: c.GameProvider, gameID: c.GameID, gameName: c.GameName, + } + addUserID(groups, key, c.UserID) +} + +func addUserID[K comparable](groups map[K]map[int64]struct{}, key K, userID int64) { + users := groups[key] + if users == nil { + users = make(map[int64]struct{}) + groups[key] = users + } + users[userID] = struct{}{} +} + +func windowDates(window dashboard.PeriodWindow) (time.Time, time.Time) { + var startDate, endDate time.Time + if window.StartDate != nil { + startDate = *window.StartDate + } + if window.EndDate != nil { + endDate = *window.EndDate + } + return startDate, endDate +} + +func ptrOrNil(value time.Time) interface{} { + if value.IsZero() { + return nil + } + return value +} + +func userIDs(users map[int64]struct{}) []int64 { + out := make([]int64, 0, len(users)) + for userID := range users { + out = append(out, userID) + } + return out +} + +func placeholders(count, columns int) string { + groups := make([]string, count) + item := "(" + strings.TrimRight(strings.Repeat("?,", columns), ",") + ")" + for i := range groups { + groups[i] = item + } + return strings.Join(groups, ",") +} diff --git a/internal/storage/batch_sql.go b/internal/storage/batch_sql.go new file mode 100644 index 0000000..1f9c87d --- /dev/null +++ b/internal/storage/batch_sql.go @@ -0,0 +1,231 @@ +package storage + +import ( + "context" + "database/sql" + "time" +) + +func (r *MySQLRepository) upsertDailyAmountTotals(ctx context.Context, tx *sql.Tx, key dailyAmountKey, totals amountTotals) error { + _, err := tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_daily_metric ( + stat_date, date_number, sys_origin, country_code, country_name, + new_user_recharge, official_recharge, mifapay_recharge, google_recharge, + dealer_recharge, salary_exchange, lucky_gift_total_flow, lucky_gift_payout, + game_total_flow, game_payout, refreshed_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE + country_name = VALUES(country_name), + new_user_recharge = new_user_recharge + VALUES(new_user_recharge), + official_recharge = official_recharge + VALUES(official_recharge), + mifapay_recharge = mifapay_recharge + VALUES(mifapay_recharge), + google_recharge = google_recharge + VALUES(google_recharge), + dealer_recharge = dealer_recharge + VALUES(dealer_recharge), + salary_exchange = salary_exchange + VALUES(salary_exchange), + lucky_gift_total_flow = lucky_gift_total_flow + VALUES(lucky_gift_total_flow), + lucky_gift_payout = lucky_gift_payout + VALUES(lucky_gift_payout), + game_total_flow = game_total_flow + VALUES(game_total_flow), + game_payout = game_payout + VALUES(game_payout), + refreshed_at = NOW() +`, key.day, dateNumber(key.day), key.sysOrigin, key.countryCode, key.countryName, + totals.newUserRecharge, totals.officialRecharge, totals.mifapayRecharge, totals.googleRecharge, + totals.dealerRecharge, totals.salaryExchange, totals.luckyGiftTotalFlow, totals.luckyGiftPayout, + totals.gameTotalFlow, totals.gamePayout) + return err +} + +func (r *MySQLRepository) upsertPeriodAmountTotals(ctx context.Context, tx *sql.Tx, key periodAmountKey, totals amountTotals) error { + _, err := tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_period_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, new_user_recharge, official_recharge, mifapay_recharge, + google_recharge, dealer_recharge, salary_exchange, lucky_gift_total_flow, lucky_gift_payout, + game_total_flow, game_payout, refreshed_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE + period_name = VALUES(period_name), + period_start_date = VALUES(period_start_date), + period_end_date = VALUES(period_end_date), + country_name = VALUES(country_name), + new_user_recharge = new_user_recharge + VALUES(new_user_recharge), + official_recharge = official_recharge + VALUES(official_recharge), + mifapay_recharge = mifapay_recharge + VALUES(mifapay_recharge), + google_recharge = google_recharge + VALUES(google_recharge), + dealer_recharge = dealer_recharge + VALUES(dealer_recharge), + salary_exchange = salary_exchange + VALUES(salary_exchange), + lucky_gift_total_flow = lucky_gift_total_flow + VALUES(lucky_gift_total_flow), + lucky_gift_payout = lucky_gift_payout + VALUES(lucky_gift_payout), + game_total_flow = game_total_flow + VALUES(game_total_flow), + game_payout = game_payout + VALUES(game_payout), + refreshed_at = NOW() +`, key.periodType, key.periodKey, key.statTimezone, key.periodName, ptrOrNil(key.startDate), ptrOrNil(key.endDate), + key.sysOrigin, key.countryCode, key.countryName, totals.newUserRecharge, totals.officialRecharge, + totals.mifapayRecharge, totals.googleRecharge, totals.dealerRecharge, totals.salaryExchange, + totals.luckyGiftTotalFlow, totals.luckyGiftPayout, totals.gameTotalFlow, totals.gamePayout) + return err +} + +func (r *MySQLRepository) upsertGameAmountTotals(ctx context.Context, tx *sql.Tx, key gameAmountKey, totals gameAmountTotals) error { + _, err := tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_game_period_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, game_provider, game_id, game_name, + consume_amount, payout_amount, profit_amount, order_count, refreshed_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE + period_name = VALUES(period_name), + period_start_date = VALUES(period_start_date), + period_end_date = VALUES(period_end_date), + country_name = VALUES(country_name), + game_name = VALUES(game_name), + consume_amount = consume_amount + VALUES(consume_amount), + payout_amount = payout_amount + VALUES(payout_amount), + profit_amount = profit_amount + VALUES(profit_amount), + order_count = order_count + VALUES(order_count), + refreshed_at = NOW() +`, key.periodType, key.periodKey, key.statTimezone, key.periodName, ptrOrNil(key.startDate), ptrOrNil(key.endDate), + key.sysOrigin, key.countryCode, key.countryName, key.gameProvider, key.gameID, key.gameName, + totals.consume, totals.payout, totals.profit, totals.orders) + return err +} + +func (r *MySQLRepository) applyDailyUserGroup(ctx context.Context, tx *sql.Tx, key dailyUserKey, users map[int64]struct{}) error { + ids := userIDs(users) + inserted, err := bulkInsertDailyUsers(ctx, tx, key, ids) + if err != nil || inserted == 0 { + return err + } + column := dailyMetricColumn(key.metricType) + if column == "" { + return nil + } + _, err = tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_daily_metric ( + stat_date, date_number, sys_origin, country_code, country_name, refreshed_at +) VALUES (?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE `+column+` = `+column+` + ?, country_name = VALUES(country_name), refreshed_at = NOW() +`, key.day, dateNumber(key.day), key.sysOrigin, key.countryCode, key.countryName, inserted) + return err +} + +func (r *MySQLRepository) applyPeriodUserGroup(ctx context.Context, tx *sql.Tx, key periodUserKey, users map[int64]struct{}) error { + ids := userIDs(users) + inserted, err := bulkInsertPeriodUsers(ctx, tx, key, ids) + if err != nil || inserted == 0 { + return err + } + column := periodMetricColumn(key.metricType) + if column == "" { + return nil + } + _, err = tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_period_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, refreshed_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE `+column+` = `+column+` + ?, country_name = VALUES(country_name), refreshed_at = NOW() +`, key.periodType, key.periodKey, key.statTimezone, key.periodName, ptrOrNil(key.startDate), ptrOrNil(key.endDate), + key.sysOrigin, key.countryCode, key.countryName, inserted) + return err +} + +func (r *MySQLRepository) applyGameUserGroup(ctx context.Context, tx *sql.Tx, key gameUserKey, users map[int64]struct{}) error { + ids := userIDs(users) + inserted, err := bulkInsertGameUsers(ctx, tx, key, ids) + if err != nil || inserted == 0 { + return err + } + _, err = tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_game_period_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, game_provider, game_id, game_name, user_count, refreshed_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE user_count = user_count + VALUES(user_count), country_name = VALUES(country_name), + game_name = VALUES(game_name), refreshed_at = NOW() +`, key.periodType, key.periodKey, key.statTimezone, key.periodName, ptrOrNil(key.startDate), ptrOrNil(key.endDate), + key.sysOrigin, key.countryCode, key.countryName, key.gameProvider, key.gameID, key.gameName, inserted) + return err +} + +func bulkInsertDailyUsers(ctx context.Context, tx *sql.Tx, key dailyUserKey, ids []int64) (int64, error) { + var inserted int64 + for start := 0; start < len(ids); start += bulkInsertChunkSize { + end := min(start+bulkInsertChunkSize, len(ids)) + args := make([]interface{}, 0, (end-start)*6) + for _, userID := range ids[start:end] { + args = append(args, key.day, key.sysOrigin, key.countryCode, key.countryName, userID, key.metricType) + } + result, err := tx.ExecContext(ctx, ` +INSERT IGNORE INTO country_dashboard_daily_user_metric ( + stat_date, sys_origin, country_code, country_name, user_id, metric_type +) VALUES `+placeholders(end-start, 6), args...) + if err != nil { + return 0, err + } + affected, err := result.RowsAffected() + if err != nil { + return 0, err + } + inserted += affected + } + return inserted, nil +} + +func bulkInsertPeriodUsers(ctx context.Context, tx *sql.Tx, key periodUserKey, ids []int64) (int64, error) { + var inserted int64 + for start := 0; start < len(ids); start += bulkInsertChunkSize { + end := min(start+bulkInsertChunkSize, len(ids)) + args := make([]interface{}, 0, (end-start)*11) + for _, userID := range ids[start:end] { + args = append(args, key.periodType, key.periodKey, key.statTimezone, key.periodName, + ptrOrNil(key.startDate), ptrOrNil(key.endDate), key.sysOrigin, key.countryCode, + key.countryName, userID, key.metricType) + } + result, err := tx.ExecContext(ctx, ` +INSERT IGNORE INTO country_dashboard_period_user_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, user_id, metric_type +) VALUES `+placeholders(end-start, 11), args...) + if err != nil { + return 0, err + } + affected, err := result.RowsAffected() + if err != nil { + return 0, err + } + inserted += affected + } + return inserted, nil +} + +func bulkInsertGameUsers(ctx context.Context, tx *sql.Tx, key gameUserKey, ids []int64) (int64, error) { + var inserted int64 + for start := 0; start < len(ids); start += bulkInsertChunkSize { + end := min(start+bulkInsertChunkSize, len(ids)) + args := make([]interface{}, 0, (end-start)*12) + for _, userID := range ids[start:end] { + args = append(args, key.periodType, key.periodKey, key.statTimezone, key.periodName, + ptrOrNil(key.startDate), ptrOrNil(key.endDate), key.sysOrigin, key.countryCode, + key.countryName, key.gameProvider, key.gameID, userID) + } + result, err := tx.ExecContext(ctx, ` +INSERT IGNORE INTO country_dashboard_game_period_user_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, game_provider, game_id, user_id +) VALUES `+placeholders(end-start, 12), args...) + if err != nil { + return 0, err + } + affected, err := result.RowsAffected() + if err != nil { + return 0, err + } + inserted += affected + } + return inserted, nil +} + +func dateNumber(day time.Time) int { + year, month, date := day.Date() + return year*10000 + int(month)*100 + date +} diff --git a/internal/storage/cache.go b/internal/storage/cache.go new file mode 100644 index 0000000..707299c --- /dev/null +++ b/internal/storage/cache.go @@ -0,0 +1,58 @@ +package storage + +import ( + "sync" + "time" +) + +type ttlCache[K comparable, V any] struct { + mu sync.RWMutex + values map[K]ttlCacheEntry[V] + ttl time.Duration + maxSize int +} + +type ttlCacheEntry[V any] struct { + value V + expiresAt time.Time +} + +func newTTLCache[K comparable, V any](ttl time.Duration, maxSize int) *ttlCache[K, V] { + return &ttlCache[K, V]{ + values: make(map[K]ttlCacheEntry[V]), + ttl: ttl, + maxSize: maxSize, + } +} + +func (c *ttlCache[K, V]) Get(key K) (V, bool) { + c.mu.RLock() + entry, ok := c.values[key] + c.mu.RUnlock() + var zero V + if !ok { + return zero, false + } + if time.Now().After(entry.expiresAt) { + c.mu.Lock() + delete(c.values, key) + c.mu.Unlock() + return zero, false + } + return entry.value, true +} + +func (c *ttlCache[K, V]) Set(key K, value V) { + if c.ttl <= 0 { + return + } + c.mu.Lock() + defer c.mu.Unlock() + if c.maxSize > 0 && len(c.values) >= c.maxSize { + c.values = make(map[K]ttlCacheEntry[V]) + } + c.values[key] = ttlCacheEntry[V]{ + value: value, + expiresAt: time.Now().Add(c.ttl), + } +} diff --git a/internal/storage/cleanup.go b/internal/storage/cleanup.go new file mode 100644 index 0000000..12a67e6 --- /dev/null +++ b/internal/storage/cleanup.go @@ -0,0 +1,119 @@ +package storage + +import ( + "context" + "log/slog" + "time" +) + +func (r *MySQLRepository) StartCleanup(ctx context.Context, logger *slog.Logger) { + if r.cfg.Dashboard.CleanupInterval <= 0 { + return + } + go func() { + r.cleanupOnce(ctx, logger) + ticker := time.NewTicker(r.cfg.Dashboard.CleanupInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + r.cleanupOnce(ctx, logger) + } + } + }() +} + +func (r *MySQLRepository) cleanupOnce(ctx context.Context, logger *slog.Logger) { + rows, err := r.Cleanup(ctx) + if err != nil { + if r.metrics != nil { + r.metrics.RecordError(err) + } + logger.Error("dashboard cleanup failed", "error", err) + return + } + if r.metrics != nil { + r.metrics.RecordCleanup(rows) + } + if rows > 0 { + logger.Info("dashboard cleanup finished", "rows", rows) + } +} + +func (r *MySQLRepository) Cleanup(ctx context.Context) (int64, error) { + var total int64 + if r.cfg.Dashboard.DedupRetention > 0 { + cutoff := time.Now().Add(-r.cfg.Dashboard.DedupRetention) + rows, err := r.deleteLimit(ctx, ` +DELETE FROM dashboard_cdc_event_dedup +WHERE created_at < ? +LIMIT ?`, cutoff, r.cfg.Dashboard.CleanupBatchSize) + if err != nil { + return total, err + } + total += rows + } + + if r.cfg.Dashboard.UserMetricDays > 0 { + cutoffDate := time.Now().AddDate(0, 0, -r.cfg.Dashboard.UserMetricDays).Format("2006-01-02") + deletes := []struct { + sql string + args []interface{} + }{ + { + sql: ` +DELETE FROM country_dashboard_daily_user_metric +WHERE stat_date < ? +LIMIT ?`, + args: []interface{}{cutoffDate, r.cfg.Dashboard.CleanupBatchSize}, + }, + { + sql: ` +DELETE FROM country_dashboard_period_user_metric +WHERE period_type <> 'ALL' + AND period_end_date IS NOT NULL + AND period_end_date < ? +LIMIT ?`, + args: []interface{}{cutoffDate, r.cfg.Dashboard.CleanupBatchSize}, + }, + { + sql: ` +DELETE FROM country_dashboard_game_period_user_metric +WHERE period_type <> 'ALL' + AND period_end_date IS NOT NULL + AND period_end_date < ? +LIMIT ?`, + args: []interface{}{cutoffDate, r.cfg.Dashboard.CleanupBatchSize}, + }, + { + sql: ` +DELETE FROM country_dashboard_recharge_detail +WHERE create_time < ? +LIMIT ?`, + args: []interface{}{cutoffDate, r.cfg.Dashboard.CleanupBatchSize}, + }, + } + for _, item := range deletes { + rows, err := r.execRows(ctx, item.sql, item.args...) + if err != nil { + return total, err + } + total += rows + } + } + return total, nil +} + +func (r *MySQLRepository) deleteLimit(ctx context.Context, query string, args ...interface{}) (int64, error) { + return r.execRows(ctx, query, args...) +} + +func (r *MySQLRepository) execRows(ctx context.Context, query string, args ...interface{}) (int64, error) { + result, err := r.db.ExecContext(ctx, query, args...) + if err != nil { + return 0, err + } + return result.RowsAffected() +} diff --git a/internal/storage/game.go b/internal/storage/game.go index 6785b76..1437b2b 100644 --- a/internal/storage/game.go +++ b/internal/storage/game.go @@ -8,12 +8,19 @@ import ( ) func (r *MySQLRepository) ResolveWalletGame(ctx context.Context, sysOrigin string, eventType string, eventID string) (string, string, string, error) { - if provider, gameID, gameName, ok, err := r.resolveBaishunByWalletEvent(ctx, sysOrigin, eventID); ok || err != nil { + if provider, gameID, fallback, ok := directGameMeta(eventType); ok { + gameName, err := r.lookupGameName(ctx, sysOrigin, provider, gameID) + if gameName == "" { + gameName = fallback + } return provider, gameID, gameName, err } if provider, gameID, gameName, ok, err := r.resolveGameOpenByWalletEvent(ctx, sysOrigin, eventType, eventID); ok || err != nil { return provider, gameID, gameName, err } + if provider, gameID, gameName, ok, err := r.resolveBaishunByWalletEvent(ctx, sysOrigin, eventID); ok || err != nil { + return provider, gameID, gameName, err + } provider, gameID, fallback := fallbackGameMeta(eventType) gameName, err := r.lookupGameName(ctx, sysOrigin, provider, gameID) @@ -87,6 +94,10 @@ LIMIT 1 } func (r *MySQLRepository) lookupGameName(ctx context.Context, sysOrigin string, provider string, gameID string) (string, error) { + cacheKey := strings.Join([]string{sysOrigin, provider, gameID}, "|") + if name, ok := r.gameNameCache.Get(cacheKey); ok { + return name, nil + } var name string if provider == "BAISHUN" { err := r.db.QueryRowContext(ctx, ` @@ -96,6 +107,9 @@ WHERE sys_origin = ? AND profile = 'PROD' AND vendor_game_id = ? LIMIT 1 `, sysOrigin, gameID).Scan(&name) if err == nil || err != sql.ErrNoRows { + if err == nil { + r.gameNameCache.Set(cacheKey, name) + } return name, err } } @@ -106,6 +120,9 @@ WHERE sys_origin = ? AND profile = 'PROD' AND vendor_game_id = ? LIMIT 1 `, sysOrigin, gameID).Scan(&name) if err == nil || err != sql.ErrNoRows { + if err == nil { + r.gameNameCache.Set(cacheKey, name) + } return name, err } err = r.db.QueryRowContext(ctx, ` @@ -115,11 +132,33 @@ WHERE sys_origin = ? AND game_id = ? LIMIT 1 `, sysOrigin, gameID).Scan(&name) if err == sql.ErrNoRows { + r.gameNameCache.Set(cacheKey, "") return "", nil } + if err == nil { + r.gameNameCache.Set(cacheKey, name) + } return name, err } +func directGameMeta(eventType string) (string, string, string, bool) { + switch { + case strings.HasPrefix(eventType, "BAISHUN_GAME_"), + strings.HasPrefix(eventType, "GAME_OPEN_"), + strings.HasPrefix(eventType, "YOMI_GAME_"), + strings.HasPrefix(eventType, "HKYS_GAME_"), + strings.HasPrefix(eventType, "HOT_GAME_"), + in(eventType, "FRUIT_GAME", "DOUBLE_LAYER_FRUIT", "GAME_FRUIT_BET", "GAME_FRUIT_AWARD", + "LUDO_GAME", "LUDO_VICTORY", "LUDO_REFUND", "GAME_LUDO_REFUND", + "GAME_BARBECUE", "DOUBLE_LAYER_BARBECUE", "GAME_SLOT_MACHINE", "TURNTABLE_GAME", + "GAME_BURST_CRYSTAL", "TEEN_PATTI", "BET_TEEN_PATTI", "LOTTERY_NUMBER_BET", "LOTTERY_REWARD"): + provider, gameID, fallback := fallbackGameMeta(eventType) + return provider, gameID, fallback, true + default: + return "", "", "", false + } +} + func fallbackGameMeta(eventType string) (string, string, string) { switch { case strings.HasPrefix(eventType, "BAISHUN_GAME_"): diff --git a/internal/storage/recharge_detail.go b/internal/storage/recharge_detail.go new file mode 100644 index 0000000..293a9cd --- /dev/null +++ b/internal/storage/recharge_detail.go @@ -0,0 +1,45 @@ +package storage + +import ( + "context" + "database/sql" + + "dashboard-cdc-worker/internal/dashboard" +) + +func (r *MySQLRepository) applyRechargeDetail(ctx context.Context, tx *sql.Tx, c dashboard.Contribution) error { + detail := c.RechargeDetail + if detail == nil || detail.RecordID == "" || detail.SourceType == "" { + return nil + } + if c.DeltaSign < 0 { + _, err := tx.ExecContext(ctx, ` +DELETE FROM country_dashboard_recharge_detail +WHERE source_type = ? AND record_id = ? +`, detail.SourceType, detail.RecordID) + return err + } + if detail.EventTime.IsZero() { + return nil + } + _, err := tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_recharge_detail ( + source_type, record_id, source_channel, sys_origin, user_id, country_code, country_name, + amount, coin_quantity, remark, create_time, update_time +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE + source_channel = VALUES(source_channel), + sys_origin = VALUES(sys_origin), + user_id = VALUES(user_id), + country_code = VALUES(country_code), + country_name = VALUES(country_name), + amount = VALUES(amount), + coin_quantity = VALUES(coin_quantity), + remark = VALUES(remark), + create_time = VALUES(create_time), + update_time = NOW() +`, detail.SourceType, detail.RecordID, detail.SourceChannel, detail.SysOrigin, detail.UserID, + detail.Country.Code, detail.Country.Name, detail.Amount, detail.CoinQuantity, + detail.Remark, detail.EventTime) + return err +} diff --git a/internal/storage/repository.go b/internal/storage/repository.go index a38c30b..3d314ca 100644 --- a/internal/storage/repository.go +++ b/internal/storage/repository.go @@ -3,34 +3,108 @@ package storage import ( "context" "database/sql" + "errors" "fmt" + "math" + "time" "github.com/go-mysql-org/go-mysql/mysql" + driverMysql "github.com/go-sql-driver/mysql" "dashboard-cdc-worker/internal/config" + "dashboard-cdc-worker/internal/dashboard" + "dashboard-cdc-worker/internal/monitor" ) const serviceName = "dashboard-cdc-worker" +const dashboardBackfillLockName = "country_dashboard_backfill_lock" type MySQLRepository struct { - db *sql.DB - cfg config.Config + db *sql.DB + cfg config.Config + metrics *monitor.Metrics + userCache *ttlCache[int64, dashboard.UserCountry] + gameNameCache *ttlCache[string, string] } -func NewMySQLRepository(db *sql.DB, cfg config.Config) *MySQLRepository { - return &MySQLRepository{db: db, cfg: cfg} +func NewMySQLRepository(db *sql.DB, cfg config.Config, metrics *monitor.Metrics) *MySQLRepository { + return &MySQLRepository{ + db: db, + cfg: cfg, + metrics: metrics, + userCache: newTTLCache[int64, dashboard.UserCountry](30*time.Minute, 200000), + gameNameCache: newTTLCache[string, string](6*time.Hour, 10000), + } } func (r *MySQLRepository) WithTx(ctx context.Context, fn func(*sql.Tx) error) error { - tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + const maxAttempts = 6 + var lastErr error + for attempt := 1; attempt <= maxAttempts; attempt++ { + tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + return err + } + err = fn(tx) + if err != nil { + _ = tx.Rollback() + } else { + err = tx.Commit() + } + if err == nil { + return nil + } + lastErr = err + if !isRetryableMySQLError(err) || attempt == maxAttempts { + return err + } + if r.metrics != nil { + r.metrics.RecordRetry() + } + backoff := time.Duration(math.Pow(2, float64(attempt-1))) * 100 * time.Millisecond + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff): + } + } + return lastErr +} + +func (r *MySQLRepository) WithDashboardLock(ctx context.Context, timeout time.Duration, fn func() error) error { + conn, err := r.db.Conn(ctx) if err != nil { return err } - if err := fn(tx); err != nil { - _ = tx.Rollback() + defer conn.Close() + + timeoutSeconds := int(math.Ceil(timeout.Seconds())) + if timeoutSeconds < 0 { + timeoutSeconds = 0 + } + var acquired sql.NullInt64 + if err := conn.QueryRowContext(ctx, `SELECT GET_LOCK(?, ?)`, dashboardBackfillLockName, timeoutSeconds).Scan(&acquired); err != nil { return err } - return tx.Commit() + if !acquired.Valid || acquired.Int64 != 1 { + if r.metrics != nil { + r.metrics.RecordLockWait() + } + return dashboard.ErrDashboardBackfillLocked + } + defer func() { + var released sql.NullInt64 + _ = conn.QueryRowContext(context.Background(), `SELECT RELEASE_LOCK(?)`, dashboardBackfillLockName).Scan(&released) + }() + return fn() +} + +func isRetryableMySQLError(err error) bool { + var mysqlErr *driverMysql.MySQLError + if !errors.As(err, &mysqlErr) { + return false + } + return mysqlErr.Number == 1213 || mysqlErr.Number == 1205 } func (r *MySQLRepository) LoadCheckpoint(ctx context.Context) (mysql.Position, bool, error) { @@ -47,7 +121,11 @@ WHERE service_name = ? if err != nil { return mysql.Position{}, false, err } - return mysql.Position{Name: file, Pos: pos}, true, nil + position := mysql.Position{Name: file, Pos: pos} + if r.metrics != nil { + r.metrics.SetCheckpointPosition(position) + } + return position, true, nil } func (r *MySQLRepository) SaveCheckpoint(ctx context.Context, pos mysql.Position) error { @@ -59,6 +137,9 @@ ON DUPLICATE KEY UPDATE binlog_pos = VALUES(binlog_pos), updated_at = NOW() `, serviceName, pos.Name, pos.Pos) + if err == nil && r.metrics != nil { + r.metrics.SetCheckpointPosition(pos) + } return err } diff --git a/internal/storage/schema.go b/internal/storage/schema.go index b7b42ce..24860fe 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -36,6 +36,7 @@ func (r *MySQLRepository) EnsureSchema(ctx context.Context) error { createPeriodUserMetricTable, createGamePeriodMetricTable, createGamePeriodUserMetricTable, + createRechargeDetailTable, }) } @@ -172,3 +173,23 @@ const createGamePeriodUserMetricTable = `CREATE TABLE IF NOT EXISTS country_dash UNIQUE KEY uk_game_period_user (period_type, period_key, stat_timezone, sys_origin, country_code, game_provider, game_id, user_id), KEY idx_game_user (sys_origin, stat_timezone, user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci` + +const createRechargeDetailTable = `CREATE TABLE IF NOT EXISTS country_dashboard_recharge_detail ( + id bigint NOT NULL AUTO_INCREMENT, + source_type varchar(32) NOT NULL, + record_id varchar(128) NOT NULL, + source_channel varchar(64) NOT NULL DEFAULT '', + sys_origin varchar(50) NOT NULL, + user_id bigint NOT NULL, + country_code varchar(64) NOT NULL DEFAULT 'UNKNOWN', + country_name varchar(128) NOT NULL DEFAULT 'UNKNOWN', + amount decimal(20,2) NOT NULL DEFAULT 0, + coin_quantity decimal(20,2) NULL, + remark varchar(512) NOT NULL DEFAULT '', + create_time datetime NOT NULL, + update_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uk_recharge_source_record (source_type, record_id), + KEY idx_recharge_query (sys_origin, create_time, country_code), + KEY idx_recharge_user (user_id, create_time) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci` diff --git a/internal/storage/user.go b/internal/storage/user.go index a95332d..7d52fc4 100644 --- a/internal/storage/user.go +++ b/internal/storage/user.go @@ -22,10 +22,16 @@ ON DUPLICATE KEY UPDATE user_updated_at = VALUES(user_updated_at), updated_at = NOW() `, user.UserID, user.SysOrigin, user.Country.Code, user.Country.Name, user.IsDeleted, nullTime(user.CreatedAt), nullTime(user.UpdatedAt)) + if err == nil { + r.userCache.Set(user.UserID, user) + } return err } func (r *MySQLRepository) UserCountry(ctx context.Context, userID int64) (dashboard.UserCountry, bool, error) { + if user, ok := r.userCache.Get(userID); ok { + return user, true, nil + } var user dashboard.UserCountry var countryCode, countryName string var isDeleted bool @@ -49,6 +55,7 @@ WHERE user_id = ? if updatedAt.Valid { user.UpdatedAt = updatedAt.Time } + r.userCache.Set(userID, user) return user, true, nil } @@ -84,6 +91,7 @@ INSERT INTO dashboard_cdc_user_country ( ) VALUES (?, ?, ?, ?, ?, ?, ?, NOW()) ON DUPLICATE KEY UPDATE updated_at = updated_at `, user.UserID, user.SysOrigin, user.Country.Code, user.Country.Name, user.IsDeleted, nullTime(user.CreatedAt), nullTime(user.UpdatedAt)) + r.userCache.Set(userID, user) return user, true, nil } diff --git a/migrations/001_dashboard_cdc_worker.sql b/migrations/001_dashboard_cdc_worker.sql index 45bd057..10c10d9 100644 --- a/migrations/001_dashboard_cdc_worker.sql +++ b/migrations/001_dashboard_cdc_worker.sql @@ -40,3 +40,23 @@ CREATE TABLE IF NOT EXISTS country_dashboard_daily_user_metric ( UNIQUE KEY uk_daily_user_metric (stat_date, sys_origin, country_code, user_id, metric_type), KEY idx_daily_user_metric (sys_origin, user_id, metric_type) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS country_dashboard_recharge_detail ( + id bigint NOT NULL AUTO_INCREMENT, + source_type varchar(32) NOT NULL, + record_id varchar(128) NOT NULL, + source_channel varchar(64) NOT NULL DEFAULT '', + sys_origin varchar(50) NOT NULL, + user_id bigint NOT NULL, + country_code varchar(64) NOT NULL DEFAULT 'UNKNOWN', + country_name varchar(128) NOT NULL DEFAULT 'UNKNOWN', + amount decimal(20,2) NOT NULL DEFAULT 0, + coin_quantity decimal(20,2) NULL, + remark varchar(512) NOT NULL DEFAULT '', + create_time datetime NOT NULL, + update_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uk_recharge_source_record (source_type, record_id), + KEY idx_recharge_query (sys_origin, create_time, country_code), + KEY idx_recharge_user (user_id, create_time) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;