From c02d057c80f62068210d685d8cbf784903ff9269 Mon Sep 17 00:00:00 2001 From: zhx Date: Wed, 27 May 2026 14:35:57 +0800 Subject: [PATCH] Reconcile gift metrics from source tables --- cmd/dashboard-cdc-worker/main.go | 5 +- internal/storage/gift_helpers.go | 183 +++++++++++++++++++ internal/storage/gift_mongo.go | 80 +++++++++ internal/storage/gift_reconcile.go | 278 +++++++++-------------------- internal/storage/gift_sql.go | 193 ++++++++++++++++++++ 5 files changed, 543 insertions(+), 196 deletions(-) create mode 100644 internal/storage/gift_helpers.go create mode 100644 internal/storage/gift_mongo.go create mode 100644 internal/storage/gift_sql.go diff --git a/cmd/dashboard-cdc-worker/main.go b/cmd/dashboard-cdc-worker/main.go index 0ae7a58..d6b5a4d 100644 --- a/cmd/dashboard-cdc-worker/main.go +++ b/cmd/dashboard-cdc-worker/main.go @@ -84,7 +84,6 @@ func main() { projector := dashboard.NewProjector(cfg, repo, redisClient, metrics) repo.StartUserReconciler(ctx, logger, projector) repo.StartRechargeReconciler(ctx, logger, projector) - repo.StartGiftReconciler(ctx, logger, projector) if cfg.Mongo.Enabled { mongoClient, err := mongo.Connect(options.Client().ApplyURI(cfg.Mongo.URI)) @@ -101,7 +100,9 @@ func main() { logger.Error("ping mongo failed", "error", err) os.Exit(1) } - repo.StartActiveReconciler(ctx, mongoClient.Database(cfg.Mongo.Database), logger) + mongoDB := mongoClient.Database(cfg.Mongo.Database) + repo.StartGiftReconciler(ctx, mongoDB, logger) + repo.StartActiveReconciler(ctx, mongoDB, logger) } worker, err := cdc.NewWorker(cfg, repo, projector, logger, metrics) diff --git a/internal/storage/gift_helpers.go b/internal/storage/gift_helpers.go new file mode 100644 index 0000000..0f5023e --- /dev/null +++ b/internal/storage/gift_helpers.go @@ -0,0 +1,183 @@ +package storage + +import ( + "fmt" + "strconv" + "strings" + "time" + + "github.com/shopspring/decimal" + "go.mongodb.org/mongo-driver/v2/bson" + + "dashboard-cdc-worker/internal/dashboard" +) + +func mergeMongoGiftRows(rows map[string]*giftMetricRow, metrics []mongoGiftAmount, anchorShare bool) { + for _, item := range metrics { + key := giftMetricKey(item.day, "", item.userID) + row := rows[key] + if row == nil { + row = &giftMetricRow{day: item.day, luckyGiftUsers: make(map[int64]struct{})} + rows[key] = row + } + if anchorShare { + row.luckyGiftAnchorShare = row.luckyGiftAnchorShare.Add(item.amount) + } else { + row.giftConsume = row.giftConsume.Add(item.amount) + } + } +} + +func mergeSQLLuckyGiftRows(rows map[string]*giftMetricRow, metrics []sqlLuckyGiftRow) { + for _, item := range metrics { + key := giftMetricKey(item.day, item.country.Code, 0) + row := rows[key] + if row == nil { + row = &giftMetricRow{ + day: item.day, + sysOrigin: item.sysOrigin, + country: item.country, + luckyGiftUsers: make(map[int64]struct{}), + } + rows[key] = row + } + row.sysOrigin = item.sysOrigin + row.country = item.country + row.luckyGiftTotalFlow = row.luckyGiftTotalFlow.Add(item.payAmount) + row.luckyGiftPayout = row.luckyGiftPayout.Add(item.payout) + row.luckyGiftUsers[item.userID] = struct{}{} + } +} + +func applyGiftUserCountries(rows map[string]*giftMetricRow, users map[int64]dashboard.UserCountry, unknownCountry string) { + for key, row := range rows { + if row.country.Code != "" { + continue + } + parts := strings.Split(key, "|") + if len(parts) != 3 { + continue + } + userID := parseInt64(parts[2]) + user, ok := users[userID] + if !ok { + delete(rows, key) + continue + } + row.sysOrigin = user.SysOrigin + row.country = user.Country + countryKey := giftMetricKey(row.day, row.country.Code, 0) + countryRow := rows[countryKey] + if countryRow == nil { + countryRow = &giftMetricRow{ + day: row.day, + sysOrigin: row.sysOrigin, + country: row.country, + luckyGiftUsers: make(map[int64]struct{}), + } + rows[countryKey] = countryRow + } + countryRow.giftConsume = countryRow.giftConsume.Add(row.giftConsume) + countryRow.luckyGiftAnchorShare = countryRow.luckyGiftAnchorShare.Add(row.luckyGiftAnchorShare) + delete(rows, key) + } +} + +func (row *giftMetricRow) empty() bool { + return row.giftConsume.IsZero() && + row.luckyGiftAnchorShare.IsZero() && + row.luckyGiftTotalFlow.IsZero() && + row.luckyGiftPayout.IsZero() && + len(row.luckyGiftUsers) == 0 +} + +func giftMetricKey(day, countryCode string, userID int64) string { + return fmt.Sprintf("%s|%s|%d", day, countryCode, userID) +} + +func int64SetToSlice(values map[int64]struct{}) []int64 { + out := make([]int64, 0, len(values)) + for value := range values { + out = append(out, value) + } + return out +} + +func bsonInt64(value interface{}) int64 { + switch typed := value.(type) { + case int64: + return typed + case int32: + return int64(typed) + case int: + return int64(typed) + case float64: + return int64(typed) + case string: + return parseInt64(typed) + default: + return parseInt64(fmt.Sprint(typed)) + } +} + +func bsonDecimal(value interface{}) decimal.Decimal { + switch typed := value.(type) { + case decimal.Decimal: + return typed + case bson.Decimal128: + parsed, err := decimal.NewFromString(typed.String()) + if err == nil { + return parsed + } + case int64: + return decimal.NewFromInt(typed) + case int32: + return decimal.NewFromInt(int64(typed)) + case int: + return decimal.NewFromInt(int64(typed)) + case float64: + return decimal.NewFromFloat(typed) + case string: + parsed, err := decimal.NewFromString(typed) + if err == nil { + return parsed + } + } + parsed, err := decimal.NewFromString(fmt.Sprint(value)) + if err != nil { + return decimal.Zero + } + return parsed +} + +func parseInt64(value string) int64 { + parsed, _ := strconv.ParseInt(strings.TrimSpace(value), 10, 64) + return parsed +} + +func defaultSysOrigin(values []string) string { + if len(values) == 0 { + return "" + } + return strings.TrimSpace(values[0]) +} + +func timezoneOffset(timezone string) string { + location, err := time.LoadLocation(timezone) + if err != nil { + return "+00:00" + } + _, offset := time.Now().In(location).Zone() + sign := "+" + if offset < 0 { + sign = "-" + offset = -offset + } + hours := offset / 3600 + minutes := (offset % 3600) / 60 + return fmt.Sprintf("%s%02d:%02d", sign, hours, minutes) +} + +func quoteIdent(value string) string { + return "`" + strings.ReplaceAll(value, "`", "``") + "`" +} diff --git a/internal/storage/gift_mongo.go b/internal/storage/gift_mongo.go new file mode 100644 index 0000000..34dc04b --- /dev/null +++ b/internal/storage/gift_mongo.go @@ -0,0 +1,80 @@ +package storage + +import ( + "context" + "fmt" + "strings" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" +) + +func (r *MySQLRepository) listMongoGiftAmounts(ctx context.Context, db *mongo.Database, + start, end time.Time, statTimezone string, luckyGift bool) ([]mongoGiftAmount, error) { + filter := bson.D{ + {Key: "refunded", Value: bson.D{{Key: "$ne", Value: true}}}, + {Key: "userId", Value: bson.D{{Key: "$ne", Value: nil}}}, + {Key: "createTime", Value: bson.D{{Key: "$gte", Value: start}, {Key: "$lt", Value: end}}}, + {Key: "sysOrigin", Value: bson.D{{Key: "$in", Value: r.cfg.Dashboard.SysOrigins}}}, + } + amountPath := "$giftValue.actualAmount" + unwindAcceptUsers := false + if luckyGift { + filter = append(filter, bson.E{Key: "luckyGift", Value: true}) + amountPath = "$acceptUsers.targetAmount" + unwindAcceptUsers = true + } else { + filter = append(filter, + bson.E{Key: "luckyGift", Value: bson.D{{Key: "$ne", Value: true}}}, + bson.E{Key: "originId", Value: bson.D{{Key: "$regex", Value: "^\\d+$"}}}, + bson.E{Key: "$or", Value: bson.A{ + bson.D{{Key: "dynamicContentId", Value: bson.D{{Key: "$exists", Value: false}}}}, + bson.D{{Key: "dynamicContentId", Value: nil}}, + bson.D{{Key: "dynamicContentId", Value: ""}}, + }}, + ) + } + pipeline := mongo.Pipeline{ + bson.D{{Key: "$match", Value: filter}}, + } + if unwindAcceptUsers { + pipeline = append(pipeline, bson.D{{Key: "$unwind", Value: "$acceptUsers"}}) + } + pipeline = append(pipeline, + bson.D{{Key: "$project", Value: bson.D{ + {Key: "userId", Value: "$userId"}, + {Key: "amount", Value: amountPath}, + {Key: "day", Value: bson.D{{Key: "$dateToString", Value: bson.D{ + {Key: "format", Value: "%Y-%m-%d"}, + {Key: "date", Value: "$createTime"}, + {Key: "timezone", Value: statTimezone}, + }}}}, + }}}, + bson.D{{Key: "$group", Value: bson.D{ + {Key: "_id", Value: bson.D{{Key: "userId", Value: "$userId"}, {Key: "day", Value: "$day"}}}, + {Key: "amount", Value: bson.D{{Key: "$sum", Value: "$amount"}}}, + }}}, + ) + cursor, err := db.Collection(giftRunningWaterCollection).Aggregate(ctx, pipeline) + if err != nil { + return nil, err + } + defer cursor.Close(ctx) + var result []mongoGiftAmount + for cursor.Next(ctx) { + var doc bson.M + if err := cursor.Decode(&doc); err != nil { + return nil, err + } + idDoc, _ := doc["_id"].(bson.M) + userID := bsonInt64(idDoc["userId"]) + day := strings.TrimSpace(fmt.Sprint(idDoc["day"])) + amount := bsonDecimal(doc["amount"]) + if userID == 0 || day == "" { + continue + } + result = append(result, mongoGiftAmount{userID: userID, day: day, amount: amount}) + } + return result, cursor.Err() +} diff --git a/internal/storage/gift_reconcile.go b/internal/storage/gift_reconcile.go index 4248113..389dea5 100644 --- a/internal/storage/gift_reconcile.go +++ b/internal/storage/gift_reconcile.go @@ -3,100 +3,59 @@ package storage import ( "context" "database/sql" - "fmt" "log/slog" - "strings" "time" "github.com/shopspring/decimal" + "go.mongodb.org/mongo-driver/v2/mongo" "dashboard-cdc-worker/internal/dashboard" ) -type walletGiftRow struct { - table string - id string - idNumber int64 - userID int64 - sysOrigin string - eventType string - recordType int64 - amount decimal.Decimal - eventTime time.Time - user dashboard.UserCountry +const giftRunningWaterCollection = "gift_give_running_water" + +type giftMetricRow struct { + day string + sysOrigin string + country dashboard.Country + giftConsume decimal.Decimal + luckyGiftAnchorShare decimal.Decimal + luckyGiftTotalFlow decimal.Decimal + luckyGiftPayout decimal.Decimal + luckyGiftUsers map[int64]struct{} } -func (r *MySQLRepository) BuildGiftReconcileEvents(ctx context.Context, since time.Time, - limit int) ([]dashboard.Event, error) { - if limit <= 0 { - limit = 50000 - } - rows, err := r.listWalletGiftRows(ctx, since, limit) - if err != nil { - return nil, err - } - events := make([]dashboard.Event, 0, len(rows)) - for _, row := range rows { - if !r.allowedOrigin(row.sysOrigin) || row.user.IsDeleted || row.user.SysOrigin != row.sysOrigin { - continue - } - contribution := dashboard.Contribution{ - SourceSchema: r.cfg.MySQL.WalletSchema, - SourceTable: row.table, - SourcePK: row.id, - SourceAction: "insert", - EventKey: dashboard.WalletMetricEventKey(r.cfg.MySQL.WalletSchema, row.table, row.id), - DeltaSign: 1, - SysOrigin: row.sysOrigin, - UserID: row.userID, - Country: row.user.Country, - EventTime: row.eventTime, - UserCreatedAt: row.user.CreatedAt, - } - switch { - case isGiftConsumeEvent(row.eventType) && row.recordType == 1: - contribution.GiftConsume = row.amount - default: - continue - } - events = append(events, dashboard.Event{ - EventKey: contribution.EventKey, - Source: r.cfg.MySQL.WalletSchema + "." + row.table, - Contributions: []dashboard.Contribution{contribution}, - }) - } - return events, nil +type mongoGiftAmount struct { + userID int64 + day string + amount decimal.Decimal } -func (r *MySQLRepository) StartGiftReconciler(ctx context.Context, logger *slog.Logger, - projector interface { - ApplyEvents(context.Context, []dashboard.Event) error - }) { +type sqlLuckyGiftRow struct { + id string + userID int64 + day string + sysOrigin string + country dashboard.Country + payAmount decimal.Decimal + payout decimal.Decimal +} + +func (r *MySQLRepository) StartGiftReconciler(ctx context.Context, db *mongo.Database, logger *slog.Logger) { if !r.cfg.Gift.Enabled || r.cfg.Gift.Interval <= 0 { return } run := func() { since := currentStatDayStart(r.cfg.Dashboard.StatTimezones, r.cfg.Gift.CurrentDayLag) - events, err := r.BuildGiftReconcileEvents(ctx, since, r.cfg.Gift.BatchLimit) - if err != nil { + until := since.AddDate(0, 0, 1) + if err := r.RebuildGiftMetricsRange(ctx, db, since, until); err != nil { if r.metrics != nil { r.metrics.RecordError(err) } - logger.Error("dashboard gift reconcile query failed", "error", err) + logger.Error("dashboard gift reconcile failed", "since", since, "until", until, "error", err) return } - if len(events) == 0 { - return - } - if err := projector.ApplyEvents(ctx, events); err != nil { - r.resetGiftHighWater() - if r.metrics != nil { - r.metrics.RecordError(err) - } - logger.Error("dashboard gift reconcile apply failed", "events", len(events), "error", err) - return - } - logger.Info("dashboard gift reconcile applied", "events", len(events), "since", since) + logger.Info("dashboard gift reconcile applied", "since", since, "until", until) } go func() { run() @@ -113,132 +72,63 @@ func (r *MySQLRepository) StartGiftReconciler(ctx context.Context, logger *slog. }() } -func (r *MySQLRepository) resetGiftHighWater() { - r.giftMu.Lock() - defer r.giftMu.Unlock() - r.giftHighWater = make(map[string]int64) -} - -func (r *MySQLRepository) listWalletGiftRows(ctx context.Context, since time.Time, - limit int) ([]walletGiftRow, error) { - tables, err := r.walletRecordTables(ctx) - if err != nil { - return nil, err - } - remaining := limit - var result []walletGiftRow - for _, table := range tables { - if remaining <= 0 { - break - } - rows, err := r.listWalletGiftRowsFromTable(ctx, table, since, remaining) +func (r *MySQLRepository) RebuildGiftMetricsRange(ctx context.Context, db *mongo.Database, start, end time.Time) error { + rowsByTimezone := make(map[string]map[string]*giftMetricRow) + userIDs := make(map[int64]struct{}) + for _, statTimezone := range r.cfg.Dashboard.StatTimezones { + location, err := time.LoadLocation(statTimezone) if err != nil { - return nil, err + location = time.UTC } - result = append(result, rows...) - remaining -= len(rows) + rows := make(map[string]*giftMetricRow) + for day := start.In(location); day.Before(end); day = day.AddDate(0, 0, 1) { + dayStart := time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, location) + dayEnd := dayStart.AddDate(0, 0, 1) + giftConsume, err := r.listMongoGiftAmounts(ctx, db, dayStart, dayEnd, statTimezone, false) + if err != nil { + return err + } + for _, item := range giftConsume { + userIDs[item.userID] = struct{}{} + } + luckyAnchorShare, err := r.listMongoGiftAmounts(ctx, db, dayStart, dayEnd, statTimezone, true) + if err != nil { + return err + } + for _, item := range luckyAnchorShare { + userIDs[item.userID] = struct{}{} + } + luckyRows, err := r.listSQLLuckyGiftRows(ctx, dayStart, dayEnd, statTimezone) + if err != nil { + return err + } + for _, item := range luckyRows { + userIDs[item.userID] = struct{}{} + } + mergeMongoGiftRows(rows, giftConsume, false) + mergeMongoGiftRows(rows, luckyAnchorShare, true) + mergeSQLLuckyGiftRows(rows, luckyRows) + } + rowsByTimezone[statTimezone] = rows } - return result, nil -} - -func (r *MySQLRepository) walletRecordTables(ctx context.Context) ([]string, error) { - rows, err := r.db.QueryContext(ctx, ` -SELECT table_name -FROM information_schema.tables -WHERE table_schema = ? - AND table_name LIKE 'wallet_gold_asset_record\_%' -ORDER BY table_name`, r.cfg.MySQL.WalletSchema) + users, err := r.loadSourceUserCountriesByID(ctx, int64SetToSlice(userIDs)) if err != nil { - return nil, err + return err } - defer rows.Close() - var tables []string - for rows.Next() { - var table string - if err := rows.Scan(&table); err != nil { - return nil, err - } - tables = append(tables, table) + for _, rows := range rowsByTimezone { + applyGiftUserCountries(rows, users, r.cfg.Dashboard.UnknownCountryCode) } - return tables, rows.Err() -} - -func (r *MySQLRepository) listWalletGiftRowsFromTable(ctx context.Context, table string, - since time.Time, limit int) ([]walletGiftRow, error) { - highWaterKey := r.cfg.MySQL.WalletSchema + "." + table - r.giftMu.Lock() - highWater := r.giftHighWater[highWaterKey] - r.giftMu.Unlock() - - sinceMillis := since.UnixMilli() - query := fmt.Sprintf(` -SELECT CAST(w.id AS CHAR), w.id, w.user_id, w.sys_origin, w.event_type, w.type, w.penny_amount, - FROM_UNIXTIME(w.create_time / 1000), - 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 %s.%s w -JOIN %s.user_base_info u ON u.id = w.user_id -WHERE w.id > ? - AND w.create_time >= ? - AND w.event_type LIKE 'GIVE_GIFT%%' - AND w.event_type NOT LIKE '%%LUCKY_GIFT%%' - AND w.type = 1 -ORDER BY w.id ASC -LIMIT ?`, quoteIdent(r.cfg.MySQL.WalletSchema), quoteIdent(table), quoteIdent(r.cfg.MySQL.Schema)) - rows, err := r.db.QueryContext(ctx, query, r.cfg.Dashboard.UnknownCountryCode, - r.cfg.Dashboard.UnknownCountryCode, highWater, sinceMillis, limit) - if err != nil { - return nil, err - } - defer rows.Close() - - var result []walletGiftRow - maxID := highWater - for rows.Next() { - var item walletGiftRow - item.table = table - var penny int64 - var countryCode, countryName string - var isDeleted bool - var userCreatedAt, userUpdatedAt sql.NullTime - if err := rows.Scan(&item.id, &item.idNumber, &item.userID, &item.sysOrigin, - &item.eventType, &item.recordType, &penny, &item.eventTime, &item.user.UserID, - &item.user.SysOrigin, &countryCode, &countryName, &isDeleted, - &userCreatedAt, &userUpdatedAt); err != nil { - return nil, err - } - item.amount = decimal.NewFromInt(penny).Div(decimal.NewFromInt(100)) - 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 - } - if item.idNumber > maxID { - maxID = item.idNumber - } - result = append(result, item) - } - if err := rows.Err(); err != nil { - return nil, err - } - if maxID > highWater { - r.giftMu.Lock() - if maxID > r.giftHighWater[highWaterKey] { - r.giftHighWater[highWaterKey] = maxID - } - r.giftMu.Unlock() - } - return result, nil -} - -func isGiftConsumeEvent(eventType string) bool { - return strings.HasPrefix(eventType, "GIVE_GIFT") && !strings.Contains(eventType, "LUCKY_GIFT") -} - -func quoteIdent(value string) string { - return "`" + strings.ReplaceAll(value, "`", "``") + "`" + return r.WithDashboardLock(ctx, r.cfg.CDC.BackfillLockWait, func() error { + return r.WithTx(ctx, func(tx *sql.Tx) error { + for statTimezone, rows := range rowsByTimezone { + if err := r.replaceGiftMetricRows(ctx, tx, statTimezone, start, end, rows); err != nil { + return err + } + if err := r.rebuildAllPeriodFromDays(ctx, tx, statTimezone); err != nil { + return err + } + } + return nil + }) + }) } diff --git a/internal/storage/gift_sql.go b/internal/storage/gift_sql.go new file mode 100644 index 0000000..e2cd5c6 --- /dev/null +++ b/internal/storage/gift_sql.go @@ -0,0 +1,193 @@ +package storage + +import ( + "context" + "database/sql" + "time" + + "github.com/shopspring/decimal" + + "dashboard-cdc-worker/internal/dashboard" +) + +func (r *MySQLRepository) listSQLLuckyGiftRows(ctx context.Context, start, end time.Time, + statTimezone string) ([]sqlLuckyGiftRow, error) { + query := ` +SELECT CAST(t.id AS CHAR), t.user_id, DATE_FORMAT(CONVERT_TZ(t.create_time, '+03:00', ?), '%Y-%m-%d'), + t.sys_origin, IFNULL(NULLIF(u.country_code, ''), ?), + IFNULL(NULLIF(u.country_name, ''), IFNULL(NULLIF(u.country_code, ''), ?)), + t.pay_amount, t.award_amount +FROM ` + quoteIdent(r.cfg.MySQL.Schema) + `.game_lucky_gift_count t +JOIN ` + quoteIdent(r.cfg.MySQL.Schema) + `.user_base_info u ON u.id = t.user_id +WHERE t.create_time >= ? + AND t.create_time < ? + AND t.sys_origin IN ` + inStringPlaceholders(len(r.cfg.Dashboard.SysOrigins)) + ` + AND u.origin_sys = t.sys_origin + AND (u.is_del = 0 OR u.is_del IS NULL)` + args := []interface{}{timezoneOffset(statTimezone), r.cfg.Dashboard.UnknownCountryCode, + r.cfg.Dashboard.UnknownCountryCode, start, end} + args = append(args, stringArgs(r.cfg.Dashboard.SysOrigins)...) + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var result []sqlLuckyGiftRow + for rows.Next() { + var item sqlLuckyGiftRow + var pay, payout decimal.Decimal + if err := rows.Scan(&item.id, &item.userID, &item.day, &item.sysOrigin, &item.country.Code, + &item.country.Name, &pay, &payout); err != nil { + return nil, err + } + item.payAmount = pay + item.payout = payout + result = append(result, item) + } + return result, rows.Err() +} + +func (r *MySQLRepository) replaceGiftMetricRows(ctx context.Context, tx *sql.Tx, + statTimezone string, start, end time.Time, rows map[string]*giftMetricRow) error { + location, err := time.LoadLocation(statTimezone) + if err != nil { + location = time.UTC + } + for day := start.In(location); day.Before(end); day = day.AddDate(0, 0, 1) { + dayStart := time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, location) + dayEnd := dayStart.AddDate(0, 0, 1) + dayKey := dayStart.Format("2006-01-02") + for _, sysOrigin := range r.cfg.Dashboard.SysOrigins { + if _, err := tx.ExecContext(ctx, ` +UPDATE country_dashboard_period_metric +SET gift_consume = 0, + lucky_gift_total_flow = 0, + lucky_gift_user = 0, + lucky_gift_payout = 0, + lucky_gift_anchor_share = 0, + refreshed_at = NOW() +WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origin = ?`, + dayKey, statTimezone, sysOrigin); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, ` +DELETE FROM country_dashboard_period_user_metric +WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origin = ? + AND metric_type = ?`, dayKey, statTimezone, sysOrigin, dashboard.MetricLuckyGiftUser); err != nil { + return err + } + } + if err := r.upsertGiftMetricRows(ctx, tx, statTimezone, dayStart, dayEnd, rows); err != nil { + return err + } + } + return nil +} + +func (r *MySQLRepository) upsertGiftMetricRows(ctx context.Context, tx *sql.Tx, + statTimezone string, dayStart, dayEnd time.Time, rows map[string]*giftMetricRow) error { + dayKey := dayStart.Format("2006-01-02") + var metricRows []*giftMetricRow + for _, row := range rows { + if row.day != dayKey || row.empty() { + continue + } + if row.sysOrigin == "" { + row.sysOrigin = defaultSysOrigin(r.cfg.Dashboard.SysOrigins) + } + if !r.allowedOrigin(row.sysOrigin) { + continue + } + if row.country.Code == "" { + row.country.Code = r.cfg.Dashboard.UnknownCountryCode + } + if row.country.Name == "" { + row.country.Name = row.country.Code + } + metricRows = append(metricRows, row) + } + if len(metricRows) == 0 { + return nil + } + args := make([]interface{}, 0, len(metricRows)*14) + for _, row := range metricRows { + args = append(args, dashboard.PeriodDay, dayKey, statTimezone, dayKey, dayStart, dayEnd, + row.sysOrigin, row.country.Code, row.country.Name, row.giftConsume, + row.luckyGiftTotalFlow, len(row.luckyGiftUsers), row.luckyGiftPayout, + row.luckyGiftAnchorShare) + } + if _, err := tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_period_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, gift_consume, lucky_gift_total_flow, + lucky_gift_user, lucky_gift_payout, lucky_gift_anchor_share +) VALUES `+placeholders(len(metricRows), 14)+` +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), + gift_consume = VALUES(gift_consume), + lucky_gift_total_flow = VALUES(lucky_gift_total_flow), + lucky_gift_user = VALUES(lucky_gift_user), + lucky_gift_payout = VALUES(lucky_gift_payout), + lucky_gift_anchor_share = VALUES(lucky_gift_anchor_share), + refreshed_at = NOW()`, args...); err != nil { + return err + } + for _, row := range metricRows { + if len(row.luckyGiftUsers) == 0 { + continue + } + key := periodUserKey{ + periodType: dashboard.PeriodDay, periodKey: dayKey, statTimezone: statTimezone, + periodName: dayKey, startDate: dayStart, endDate: dayEnd, sysOrigin: row.sysOrigin, + countryCode: row.country.Code, countryName: row.country.Name, metricType: dashboard.MetricLuckyGiftUser, + } + if _, err := bulkInsertPeriodUsers(ctx, tx, key, int64SetToSlice(row.luckyGiftUsers)); err != nil { + return err + } + } + return nil +} + +func (r *MySQLRepository) loadSourceUserCountriesByID(ctx context.Context, ids []int64) (map[int64]dashboard.UserCountry, error) { + result := make(map[int64]dashboard.UserCountry) + for start := 0; start < len(ids); start += bulkInsertChunkSize { + end := min(start+bulkInsertChunkSize, len(ids)) + args := make([]interface{}, 0, end-start+2) + args = append(args, r.cfg.Dashboard.UnknownCountryCode, r.cfg.Dashboard.UnknownCountryCode) + for _, id := range ids[start:end] { + args = append(args, id) + } + rows, err := r.db.QueryContext(ctx, ` +SELECT id, origin_sys, IFNULL(NULLIF(country_code, ''), ?), + IFNULL(NULLIF(country_name, ''), IFNULL(NULLIF(country_code, ''), ?)), + IFNULL(is_del, 0), create_time, update_time +FROM `+quoteIdent(r.cfg.MySQL.Schema)+`.user_base_info +WHERE id IN `+inPlaceholders(end-start), args...) + if err != nil { + return nil, err + } + for rows.Next() { + var user dashboard.UserCountry + var createdAt, updatedAt sql.NullTime + if err := rows.Scan(&user.UserID, &user.SysOrigin, &user.Country.Code, &user.Country.Name, + &user.IsDeleted, &createdAt, &updatedAt); err != nil { + _ = rows.Close() + return nil, err + } + if createdAt.Valid { + user.CreatedAt = createdAt.Time + } + if updatedAt.Valid { + user.UpdatedAt = updatedAt.Time + } + result[user.UserID] = user + } + if err := rows.Close(); err != nil { + return nil, err + } + } + return result, nil +}