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

fix: align dashboard gift time windows

Browse Source
This commit is contained in:
zhx 2026-05-28 11:44:54 +08:00
parent fcc11122d4
commit 358036a847
8 changed files with 71 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
internal/cdc/handler.go
View File

@ -37,7 +37,8 @@ func newHandler(cfg config.Config, store dashboard.Store, projector *dashboard.P
store: store,
projector: projector,
mapper: dashboard.NewMapper(store, cfg.Dashboard.SysOrigins,
cfg.Dashboard.UnknownCountryCode, cfg.User.CreateTimeDisplayOffset),
cfg.Dashboard.UnknownCountryCode, cfg.User.CreateTimeDisplayOffset,
cfg.Dashboard.SourceTimeOffset),
logger: logger,
batch: make([]dashboard.Event, 0, cfg.CDC.BatchSize),
lastFlush: time.Now(),

2
internal/config/config.go
View File

@ -65,6 +65,7 @@ type DashboardConfig struct {
SysOrigins []string
StatTimezones []string
StorageTimezone string
SourceTimeOffset time.Duration
RedisKeyPrefix string
RealtimeRedisTTL time.Duration
UnknownCountryCode string
@ -153,6 +154,7 @@ func Load() (Config, error) {
SysOrigins: csv(env("DASHBOARD_SYS_ORIGINS", "LIKEI")),
StatTimezones: csv(env("DASHBOARD_STAT_TIMEZONES", "Asia/Riyadh")),
StorageTimezone: env("DASHBOARD_STORAGE_TIMEZONE", "Asia/Riyadh"),
SourceTimeOffset: envDuration("DASHBOARD_SOURCE_TIME_OFFSET", 8*time.Hour),
RedisKeyPrefix: env("DASHBOARD_REDIS_KEY_PREFIX", "country_dashboard"),
RealtimeRedisTTL: envDuration("DASHBOARD_REALTIME_REDIS_TTL", 48*time.Hour),
UnknownCountryCode: env("DASHBOARD_UNKNOWN_COUNTRY_CODE", "UNKNOWN"),

8
internal/dashboard/mapper.go
View File

@ -13,19 +13,25 @@ type Mapper struct {
sysOrigins map[string]struct{}
unknownCountry string
userCreateDisplayOffset time.Duration
sourceTimeOffset time.Duration
}
func NewMapper(store Store, sysOrigins []string, unknownCountry string,
userCreateDisplayOffset time.Duration) *Mapper {
userCreateDisplayOffset time.Duration, sourceTimeOffset ...time.Duration) *Mapper {
origins := make(map[string]struct{}, len(sysOrigins))
for _, origin := range sysOrigins {
origins[strings.TrimSpace(origin)] = struct{}{}
}
offset := time.Duration(0)
if len(sourceTimeOffset) > 0 {
offset = sourceTimeOffset[0]
}
return &Mapper{
store: store,
sysOrigins: origins,
unknownCountry: unknownCountry,
userCreateDisplayOffset: userCreateDisplayOffset,
sourceTimeOffset: offset,
}
}

10
internal/dashboard/mapper_helpers.go
View File

@ -29,11 +29,19 @@ func (m *Mapper) amountContribution(
if user.SysOrigin != sysOrigin {
return nil, nil
}
c := m.base(schema, table, action, row, user, row.Time("create_time"))
c := m.base(schema, table, action, row, user, m.sourceTime(row, "create_time"))
apply(&c, amount)
return []Contribution{c}, nil
}
func (m *Mapper) sourceTime(row Row, field string) time.Time {
value := row.Time(field)
if value.IsZero() || m.sourceTimeOffset == 0 {
return value
}
return value.Add(m.sourceTimeOffset)
}
func (m *Mapper) base(schema string, table string, action string, row Row, user UserCountry, eventTime time.Time) Contribution {
return Contribution{
SourceSchema: schema,

26
internal/dashboard/mapper_wallet_test.go
View File

@ -124,6 +124,32 @@ func TestUserCreateTimeDisplayOffsetIsApplied(t *testing.T) {
}
}
func TestSourceTimeOffsetIsAppliedToAmountEvents(t *testing.T) {
mapper := NewMapper(walletMapperTestStore{users: map[int64]UserCountry{
1001: {UserID: 1001, SysOrigin: "LIKEI", Country: Country{Code: "BD", Name: "Bangladesh"}},
}}, []string{"LIKEI"}, "UNKNOWN", 0, 8*time.Hour)
decodedCreateTime := time.Date(2026, 5, 27, 19, 36, 0, 0, time.UTC)
contributions, err := mapper.Map(context.Background(), "likei", "game_lucky_gift_count", "insert", Row{
"id": int64(31),
"user_id": int64(1001),
"sys_origin": "LIKEI",
"pay_amount": decimal.NewFromInt(50),
"award_amount": decimal.NewFromInt(100),
"create_time": decodedCreateTime,
})
if err != nil {
t.Fatal(err)
}
if len(contributions) != 1 {
t.Fatalf("expected one contribution, got %d", len(contributions))
}
expected := decodedCreateTime.Add(8 * time.Hour)
if !contributions[0].EventTime.Equal(expected) {
t.Fatalf("expected event time %s, got %s", expected, contributions[0].EventTime)
}
}
func TestGoogleRechargeContributesToUserRecharge(t *testing.T) {
mapper := NewMapper(walletMapperTestStore{users: map[int64]UserCountry{
1001: {

14
internal/dashboard/period.go
View File

@ -6,7 +6,7 @@ import (
)
func PeriodWindows(eventTime time.Time, statZone *time.Location) []PeriodWindow {
localDate := startOfDay(wallClockInLocation(eventTime, statZone))
localDate := startOfDay(timeInLocation(eventTime, statZone))
dayEnd := localDate.AddDate(0, 0, 1)
weekday := int(localDate.Weekday())
@ -59,8 +59,8 @@ func SamePeriod(left time.Time, right time.Time, window PeriodWindow, statZone *
if window.Type == PeriodAll {
return true
}
leftLocal := wallClockInLocation(left, statZone)
rightLocal := wallClockInLocation(right, statZone)
leftLocal := timeInLocation(left, statZone)
rightLocal := timeInLocation(right, statZone)
switch window.Type {
case PeriodDay:
return startOfDay(leftLocal).Equal(startOfDay(rightLocal))
@ -76,7 +76,7 @@ func SamePeriod(left time.Time, right time.Time, window PeriodWindow, statZone *
}
func DailyDate(eventTime time.Time, storageZone *time.Location) time.Time {
return startOfDay(wallClockInLocation(eventTime, storageZone))
return startOfDay(timeInLocation(eventTime, storageZone))
}
func DateNumber(day time.Time) int {
@ -94,14 +94,12 @@ func startOfDay(value time.Time) time.Time {
return time.Date(year, month, date, 0, 0, 0, 0, value.Location())
}
func wallClockInLocation(value time.Time, location *time.Location) time.Time {
func timeInLocation(value time.Time, location *time.Location) time.Time {
if value.IsZero() {
return value
}
if location == nil {
location = time.UTC
}
year, month, day := value.Date()
hour, minute, second := value.Clock()
return time.Date(year, month, day, hour, minute, second, value.Nanosecond(), location)
return value.In(location)
}

24
internal/dashboard/period_test.go
View File

@ -5,38 +5,38 @@ import (
"time"
)
func TestDailyDateKeepsDatabaseDatetimeCalendarDay(t *testing.T) {
func TestDailyDateUsesStatisticTimezone(t *testing.T) {
riyadh := mustLoadLocation(t, "Asia/Riyadh")
eventTime := time.Date(2026, 5, 26, 1, 30, 0, 0, time.UTC)
eventTime := time.Date(2026, 5, 26, 21, 30, 0, 0, time.UTC)
got := DailyDate(eventTime, riyadh)
if got.Format("2006-01-02") != "2026-05-26" {
t.Fatalf("DailyDate shifted database DATETIME calendar day: got %s", got.Format("2006-01-02"))
if got.Format("2006-01-02") != "2026-05-27" {
t.Fatalf("DailyDate = %s, want Riyadh day 2026-05-27", got.Format("2006-01-02"))
}
if got.Location() != riyadh {
t.Fatalf("DailyDate location = %s, want %s", got.Location(), riyadh)
}
}
func TestPeriodWindowsKeepsDatabaseDatetimeCalendarDay(t *testing.T) {
func TestPeriodWindowsUseStatisticTimezone(t *testing.T) {
riyadh := mustLoadLocation(t, "Asia/Riyadh")
eventTime := time.Date(2026, 5, 26, 1, 30, 0, 0, time.UTC)
eventTime := time.Date(2026, 5, 26, 21, 30, 0, 0, time.UTC)
windows := PeriodWindows(eventTime, riyadh)
if windows[0].Type != PeriodDay || windows[0].Key != "2026-05-26" {
t.Fatalf("day window = %+v, want key 2026-05-26", windows[0])
if windows[0].Type != PeriodDay || windows[0].Key != "2026-05-27" {
t.Fatalf("day window = %+v, want key 2026-05-27", windows[0])
}
}
func TestSamePeriodKeepsDatabaseDatetimeCalendarDay(t *testing.T) {
func TestSamePeriodUsesStatisticTimezone(t *testing.T) {
riyadh := mustLoadLocation(t, "Asia/Riyadh")
left := time.Date(2026, 5, 26, 0, 10, 0, 0, time.UTC)
right := time.Date(2026, 5, 26, 23, 50, 0, 0, time.UTC)
left := time.Date(2026, 5, 26, 21, 10, 0, 0, time.UTC)
right := time.Date(2026, 5, 27, 20, 50, 0, 0, time.UTC)
if !SamePeriod(left, right, PeriodWindow{Type: PeriodDay}, riyadh) {
t.Fatal("SamePeriod shifted database DATETIME values out of the same calendar day")
t.Fatal("SamePeriod should match events in the same Riyadh day")
}
}

12
internal/storage/gift_sql.go
View File

@ -12,8 +12,10 @@ import (
func (r *MySQLRepository) listSQLLuckyGiftRows(ctx context.Context, start, end time.Time,
statTimezone string) ([]sqlLuckyGiftRow, error) {
startUTC := start.In(time.UTC).Format("2006-01-02 15:04:05")
endUTC := end.In(time.UTC).Format("2006-01-02 15:04:05")
query := `
SELECT CAST(t.id AS CHAR), t.user_id, DATE_FORMAT(CONVERT_TZ(t.create_time, '+03:00', ?), '%Y-%m-%d'),
SELECT CAST(t.id AS CHAR), t.user_id, DATE_FORMAT(CONVERT_TZ(t.create_time, '+00: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
@ -21,11 +23,11 @@ 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)`
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}
r.cfg.Dashboard.UnknownCountryCode, startUTC, endUTC}
args = append(args, stringArgs(r.cfg.Dashboard.SysOrigins)...)
rows, err := r.db.QueryContext(ctx, query, args...)
if err != nil {