diff --git a/cmd/dashboard-cdc-worker/main.go b/cmd/dashboard-cdc-worker/main.go index 599f314..f3e617e 100644 --- a/cmd/dashboard-cdc-worker/main.go +++ b/cmd/dashboard-cdc-worker/main.go @@ -12,6 +12,8 @@ import ( _ "github.com/go-sql-driver/mysql" "github.com/redis/go-redis/v9" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" "dashboard-cdc-worker/internal/cdc" "dashboard-cdc-worker/internal/config" @@ -83,6 +85,24 @@ func main() { repo.StartRechargeReconciler(ctx, logger, projector) repo.StartGiftReconciler(ctx, logger, projector) + if cfg.Mongo.Enabled { + mongoClient, err := mongo.Connect(options.Client().ApplyURI(cfg.Mongo.URI)) + if err != nil { + logger.Error("connect mongo failed", "error", err) + os.Exit(1) + } + defer func() { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _ = mongoClient.Disconnect(shutdownCtx) + }() + if err := mongoClient.Ping(ctx, nil); err != nil { + logger.Error("ping mongo failed", "error", err) + os.Exit(1) + } + repo.StartActiveReconciler(ctx, mongoClient.Database(cfg.Mongo.Database), logger) + } + worker, err := cdc.NewWorker(cfg, repo, projector, logger, metrics) if err != nil { logger.Error("create cdc worker failed", "error", err) diff --git a/env.example b/env.example index 3d1e4fd..19c80c8 100644 --- a/env.example +++ b/env.example @@ -9,6 +9,9 @@ CDC_BATCH_SIZE=1000 CDC_FLUSH_INTERVAL=3s CDC_BACKFILL_LOCK_WAIT=1s CDC_MASTER_POSITION_POLL_INTERVAL=1m +MONGO_ENABLED=false +MONGO_URI= +MONGO_DATABASE= DASHBOARD_SYS_ORIGINS=LIKEI DASHBOARD_STAT_TIMEZONES=Asia/Riyadh DASHBOARD_STORAGE_TIMEZONE=Asia/Riyadh @@ -16,6 +19,10 @@ DASHBOARD_CLEANUP_INTERVAL=30m DASHBOARD_DEDUP_RETENTION=72h DASHBOARD_USER_METRIC_RETENTION_DAYS=120 DASHBOARD_CLEANUP_BATCH_SIZE=5000 +DASHBOARD_ACTIVE_RECONCILE_ENABLED=false +DASHBOARD_ACTIVE_RECONCILE_INTERVAL=10m +DASHBOARD_ACTIVE_RECONCILE_LOOKBACK_DAYS=35 +DASHBOARD_ACTIVE_RECONCILE_BATCH_LIMIT=200000 REDIS_ENABLED=false REDIS_ADDR=127.0.0.1:6379 HEALTH_ADDR=:2910 diff --git a/go.mod b/go.mod index 52ae970..1781158 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/go-sql-driver/mysql v1.9.3 github.com/redis/go-redis/v9 v9.12.1 github.com/shopspring/decimal v1.4.0 + go.mongodb.org/mongo-driver/v2 v2.3.1 ) require ( @@ -15,15 +16,22 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/goccy/go-json v0.10.2 // indirect + github.com/golang/snappy v1.0.0 // indirect github.com/google/uuid v1.3.0 // indirect github.com/klauspost/compress v1.17.8 // indirect github.com/pingcap/errors v0.11.5-0.20250318082626-8f80e5cb09ec // indirect github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect github.com/pingcap/log v1.1.1-0.20241212030209-7e3ff8601a2a // indirect github.com/pingcap/tidb/pkg/parser v0.0.0-20250421232622-526b2c79173d // indirect + github.com/xdg-go/pbkdf2 v1.0.0 // indirect + github.com/xdg-go/scram v1.1.2 // indirect + github.com/xdg-go/stringprep v1.0.4 // indirect + github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.27.0 // indirect + golang.org/x/crypto v0.33.0 // indirect + golang.org/x/sync v0.13.0 // indirect golang.org/x/text v0.24.0 // indirect gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect ) diff --git a/go.sum b/go.sum index c0e63ff..ef72ab3 100644 --- a/go.sum +++ b/go.sum @@ -20,6 +20,10 @@ github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1 github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU= github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU= github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I= +github.com/golang/snappy v1.0.0 h1:Oy607GVXHs7RtbggtPBnr2RmDArIsAefDwvrdWvRhGs= +github.com/golang/snappy v1.0.0/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU= @@ -49,6 +53,17 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c= +github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= +github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY= +github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4= +github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8= +github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM= +github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM= +github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +go.mongodb.org/mongo-driver/v2 v2.3.1 h1:WrCgSzO7dh1/FrePud9dK5fKNZOE97q5EQimGkos7Wo= +go.mongodb.org/mongo-driver/v2 v2.3.1/go.mod h1:jHeEDJHJq7tm6ZF45Issun9dbogjfnPySb1vXA7EeAI= go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= @@ -65,17 +80,38 @@ go.uber.org/zap v1.19.0/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI= go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.33.0 h1:IOBPskki6Lysi0lo9qQvbxiQ+FvsCC/YWOecCHAixus= +golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M= golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610= +golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0= golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/internal/config/config.go b/internal/config/config.go index 5bd76b6..b0d551a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -12,11 +12,13 @@ import ( type Config struct { MySQL MySQLConfig + Mongo MongoConfig Redis RedisConfig CDC CDCConfig Dashboard DashboardConfig Recharge RechargeReconcileConfig Gift GiftReconcileConfig + Active ActiveReconcileConfig Health HealthConfig } @@ -31,6 +33,12 @@ type MySQLConfig struct { MaxIdleConns int } +type MongoConfig struct { + Enabled bool + URI string + Database string +} + type RedisConfig struct { Enabled bool Addr string @@ -78,6 +86,13 @@ type GiftReconcileConfig struct { BatchLimit int } +type ActiveReconcileConfig struct { + Enabled bool + Interval time.Duration + LookbackDays int + BatchLimit int +} + type HealthConfig struct { Addr string } @@ -94,6 +109,11 @@ func Load() (Config, error) { MaxOpenConns: envInt("MYSQL_MAX_OPEN_CONNS", 8), MaxIdleConns: envInt("MYSQL_MAX_IDLE_CONNS", 4), }, + Mongo: MongoConfig{ + Enabled: envBool("MONGO_ENABLED", false), + URI: env("MONGO_URI", env("LIKEI_MONGO_URI", "")), + Database: env("MONGO_DATABASE", env("LIKEI_MONGO_DB", "")), + }, Redis: RedisConfig{ Enabled: envBool("REDIS_ENABLED", false), Addr: env("REDIS_ADDR", ""), @@ -136,6 +156,12 @@ func Load() (Config, error) { CurrentDayLag: envDuration("DASHBOARD_GIFT_RECONCILE_CURRENT_DAY_LAG", 5*time.Minute), BatchLimit: envInt("DASHBOARD_GIFT_RECONCILE_BATCH_LIMIT", 50000), }, + Active: ActiveReconcileConfig{ + Enabled: envBool("DASHBOARD_ACTIVE_RECONCILE_ENABLED", false), + Interval: envDuration("DASHBOARD_ACTIVE_RECONCILE_INTERVAL", 10*time.Minute), + LookbackDays: envInt("DASHBOARD_ACTIVE_RECONCILE_LOOKBACK_DAYS", 35), + BatchLimit: envInt("DASHBOARD_ACTIVE_RECONCILE_BATCH_LIMIT", 200000), + }, Health: HealthConfig{ Addr: env("HEALTH_ADDR", ":2910"), }, @@ -153,6 +179,12 @@ func Load() (Config, error) { if cfg.Redis.Enabled && cfg.Redis.Addr == "" { return Config{}, errors.New("REDIS_ADDR is required when REDIS_ENABLED=true") } + if cfg.Mongo.Enabled && cfg.Mongo.URI == "" { + return Config{}, errors.New("MONGO_URI or LIKEI_MONGO_URI is required when MONGO_ENABLED=true") + } + if cfg.Mongo.Enabled && cfg.Mongo.Database == "" { + return Config{}, errors.New("MONGO_DATABASE or LIKEI_MONGO_DB is required when MONGO_ENABLED=true") + } if cfg.CDC.ServerID == 0 { return Config{}, errors.New("CDC_SERVER_ID must be greater than 0") } @@ -168,6 +200,12 @@ func Load() (Config, error) { if cfg.Gift.BatchLimit <= 0 { cfg.Gift.BatchLimit = 50000 } + if cfg.Active.LookbackDays <= 0 { + cfg.Active.LookbackDays = 35 + } + if cfg.Active.BatchLimit <= 0 { + cfg.Active.BatchLimit = 200000 + } 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_wallet.go b/internal/dashboard/mapper_wallet.go index 3e381d1..6f5d218 100644 --- a/internal/dashboard/mapper_wallet.go +++ b/internal/dashboard/mapper_wallet.go @@ -18,7 +18,7 @@ var walletGameEvents = map[string]struct{}{ func (m *Mapper) mapWalletGame(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { eventType := row.String("event_type") - if eventType == "LUCKY_GIFT" || isWalletGiftConsumeEvent(eventType) { + if isWalletGiftConsumeEvent(eventType) { return m.mapWalletGift(ctx, schema, table, action, row) } if !isWalletGameEvent(eventType) { @@ -87,11 +87,6 @@ func (m *Mapper) mapWalletGift(ctx context.Context, schema string, table string, c.EventKey = WalletMetricEventKey(schema, table, rowPK(row)) } switch { - case eventType == "LUCKY_GIFT" && recordType == 1: - c.LuckyGiftTotalFlow = amount - c.LuckyGiftUser = true - case eventType == "LUCKY_GIFT" && recordType == 0: - c.LuckyGiftPayout = amount case isWalletGiftConsumeEvent(eventType) && recordType == 1: c.GiftConsume = amount default: diff --git a/internal/dashboard/mapper_wallet_test.go b/internal/dashboard/mapper_wallet_test.go new file mode 100644 index 0000000..122e54d --- /dev/null +++ b/internal/dashboard/mapper_wallet_test.go @@ -0,0 +1,101 @@ +package dashboard + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/go-mysql-org/go-mysql/mysql" + "github.com/shopspring/decimal" +) + +type walletMapperTestStore struct { + users map[int64]UserCountry +} + +func (s walletMapperTestStore) WithTx(context.Context, func(*sql.Tx) error) error { return nil } +func (s walletMapperTestStore) WithDashboardLock(context.Context, time.Duration, func() error) error { + return nil +} +func (s walletMapperTestStore) MarkEventApplied(context.Context, *sql.Tx, string, string) (bool, error) { + return true, nil +} +func (s walletMapperTestStore) SaveCheckpoint(context.Context, mysql.Position) error { return nil } +func (s walletMapperTestStore) LoadCheckpoint(context.Context) (mysql.Position, bool, error) { + return mysql.Position{}, false, nil +} +func (s walletMapperTestStore) Cleanup(context.Context) (int64, error) { return 0, nil } +func (s walletMapperTestStore) UpsertUserCountry(context.Context, *sql.Tx, UserCountry) error { + return nil +} +func (s walletMapperTestStore) UserCountry(_ context.Context, userID int64) (UserCountry, bool, error) { + user, ok := s.users[userID] + return user, ok, nil +} +func (s walletMapperTestStore) ResolveWalletGame(context.Context, string, string, string) (string, string, string, error) { + return "TEST", "1", "Test", nil +} +func (s walletMapperTestStore) ApplyDailyContribution(context.Context, *sql.Tx, Contribution, string) error { + return nil +} +func (s walletMapperTestStore) ApplyPeriodContribution(context.Context, *sql.Tx, Contribution, string) error { + return nil +} +func (s walletMapperTestStore) ApplyGamePeriodContribution(context.Context, *sql.Tx, Contribution, string) error { + return nil +} +func (s walletMapperTestStore) ApplyContributionBatch(context.Context, *sql.Tx, []Contribution, string, []string) error { + return nil +} + +func TestWalletLuckyGiftIsNotMappedAsGiftConsume(t *testing.T) { + mapper := NewMapper(walletMapperTestStore{users: map[int64]UserCountry{ + 1001: {UserID: 1001, SysOrigin: "LIKEI", Country: Country{Code: "BD", Name: "Bangladesh"}}, + }}, []string{"LIKEI"}, "UNKNOWN") + + row := Row{ + "id": int64(11), + "user_id": int64(1001), + "sys_origin": "LIKEI", + "event_type": "LUCKY_GIFT", + "type": int64(1), + "penny_amount": decimal.NewFromInt(10000), + "create_time": time.Now().UnixMilli(), + } + + contributions, err := mapper.Map(context.Background(), "likei_wallet", "wallet_gold_asset_record_1", "insert", row) + if err != nil { + t.Fatal(err) + } + if len(contributions) != 0 { + t.Fatalf("expected wallet LUCKY_GIFT to be ignored, got %#v", contributions) + } +} + +func TestWalletGiveGiftIsMappedAsGiftConsume(t *testing.T) { + mapper := NewMapper(walletMapperTestStore{users: map[int64]UserCountry{ + 1001: {UserID: 1001, SysOrigin: "LIKEI", Country: Country{Code: "BD", Name: "Bangladesh"}}, + }}, []string{"LIKEI"}, "UNKNOWN") + + row := Row{ + "id": int64(12), + "user_id": int64(1001), + "sys_origin": "LIKEI", + "event_type": "GIVE_GIFT_V3", + "type": int64(1), + "penny_amount": decimal.NewFromInt(10000), + "create_time": time.Now().UnixMilli(), + } + + contributions, err := mapper.Map(context.Background(), "likei_wallet", "wallet_gold_asset_record_1", "insert", row) + if err != nil { + t.Fatal(err) + } + if len(contributions) != 1 { + t.Fatalf("expected one gift contribution, got %d", len(contributions)) + } + if !contributions[0].GiftConsume.Equal(decimal.NewFromInt(100)) { + t.Fatalf("expected gift consume 100, got %s", contributions[0].GiftConsume.String()) + } +} diff --git a/internal/storage/active_reconcile.go b/internal/storage/active_reconcile.go new file mode 100644 index 0000000..2dbb846 --- /dev/null +++ b/internal/storage/active_reconcile.go @@ -0,0 +1,500 @@ +package storage + +import ( + "context" + "database/sql" + "log/slog" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" + + "dashboard-cdc-worker/internal/dashboard" +) + +const activeLogCollection = "user_daily_active_log" + +type activeLogRow struct { + UserID int64 `bson:"userId"` + ActiveDay string `bson:"activeDate"` + SysOrigin string `bson:"sysOrigin"` +} + +type activeDayStats struct { + active map[string]map[int64]struct{} +} + +type activeRetentionValues struct { + countryName string + activeUsers int64 + d1User int64 + d1Base int64 + d7User int64 + d7Base int64 + d30User int64 + d30Base int64 +} + +type countryCount struct { + name string + count int64 +} + +func (r *MySQLRepository) StartActiveReconciler(ctx context.Context, db *mongo.Database, logger *slog.Logger) { + if !r.cfg.Active.Enabled || r.cfg.Active.Interval <= 0 { + return + } + run := func() { + if err := r.RebuildRecentActiveRetention(ctx, db); err != nil { + if r.metrics != nil { + r.metrics.RecordError(err) + } + logger.Error("dashboard active reconcile failed", "error", err) + return + } + logger.Info("dashboard active reconcile applied", "lookbackDays", r.cfg.Active.LookbackDays) + } + go func() { + run() + ticker := time.NewTicker(r.cfg.Active.Interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + run() + } + } + }() +} + +func (r *MySQLRepository) RebuildRecentActiveRetention(ctx context.Context, db *mongo.Database) error { + location, err := time.LoadLocation(r.cfg.Dashboard.StorageTimezone) + if err != nil { + location = time.UTC + } + today := time.Now().In(location) + end := time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, location).AddDate(0, 0, 1) + start := end.AddDate(0, 0, -r.cfg.Active.LookbackDays) + return r.RebuildActiveRetentionRange(ctx, db, start, end) +} + +func (r *MySQLRepository) RebuildActiveRetentionRange(ctx context.Context, db *mongo.Database, start, end time.Time) error { + events, err := r.loadActiveLogs(ctx, db, start, end) + if err != nil { + return err + } + users, err := r.loadUserCountriesByID(ctx, collectActiveUserIDs(events)) + if err != nil { + return err + } + return r.WithDashboardLock(ctx, r.cfg.CDC.BackfillLockWait, func() error { + return r.WithTx(ctx, func(tx *sql.Tx) error { + for day := start; day.Before(end); day = day.AddDate(0, 0, 1) { + if err := r.replaceActiveRetentionDay(ctx, tx, day, events, users); err != nil { + return err + } + } + for _, statTimezone := range r.cfg.Dashboard.StatTimezones { + if err := r.rebuildAllPeriodFromDays(ctx, tx, statTimezone); err != nil { + return err + } + } + return nil + }) + }) +} + +func (r *MySQLRepository) loadActiveLogs(ctx context.Context, db *mongo.Database, start, end time.Time) (map[string]map[int64]struct{}, error) { + filter := bson.D{ + {Key: "activeDate", Value: bson.D{{Key: "$gte", Value: start.Format("2006-01-02")}, {Key: "$lt", Value: end.Format("2006-01-02")}}}, + {Key: "sysOrigin", Value: bson.D{{Key: "$in", Value: r.cfg.Dashboard.SysOrigins}}}, + } + findOptions := options.Find().SetProjection(bson.D{ + {Key: "userId", Value: 1}, + {Key: "activeDate", Value: 1}, + {Key: "sysOrigin", Value: 1}, + }) + if r.cfg.Active.BatchLimit > 0 { + findOptions.SetLimit(int64(r.cfg.Active.BatchLimit)) + } + cursor, err := db.Collection(activeLogCollection).Find(ctx, filter, findOptions) + if err != nil { + return nil, err + } + defer cursor.Close(ctx) + + events := make(map[string]map[int64]struct{}) + for cursor.Next(ctx) { + var row activeLogRow + if err := cursor.Decode(&row); err != nil { + return nil, err + } + if row.UserID == 0 || row.ActiveDay == "" || !r.allowedOrigin(row.SysOrigin) { + continue + } + users := events[row.ActiveDay] + if users == nil { + users = make(map[int64]struct{}) + events[row.ActiveDay] = users + } + users[row.UserID] = struct{}{} + } + return events, cursor.Err() +} + +func collectActiveUserIDs(events map[string]map[int64]struct{}) []int64 { + seen := make(map[int64]struct{}) + for _, users := range events { + for userID := range users { + seen[userID] = struct{}{} + } + } + out := make([]int64, 0, len(seen)) + for userID := range seen { + out = append(out, userID) + } + return out +} + +func (r *MySQLRepository) loadUserCountriesByID(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) + for _, id := range ids[start:end] { + args = append(args, id) + } + rows, err := r.db.QueryContext(ctx, ` +SELECT user_id, sys_origin, country_code, country_name, is_deleted, user_created_at, user_updated_at +FROM dashboard_cdc_user_country +WHERE user_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 +} + +func (r *MySQLRepository) replaceActiveRetentionDay(ctx context.Context, tx *sql.Tx, day time.Time, + events map[string]map[int64]struct{}, users map[int64]dashboard.UserCountry) error { + dayKey := day.Format("2006-01-02") + statsByOrigin := make(map[string]map[string]*activeDayStats) + for userID := range events[dayKey] { + user, ok := users[userID] + if !ok || user.IsDeleted || !r.allowedOrigin(user.SysOrigin) { + continue + } + stats := statsByOrigin[user.SysOrigin] + if stats == nil { + stats = make(map[string]*activeDayStats) + statsByOrigin[user.SysOrigin] = stats + } + stat := stats[user.Country.Code] + if stat == nil { + stat = &activeDayStats{active: make(map[string]map[int64]struct{})} + stats[user.Country.Code] = stat + } + active := stat.active[user.Country.Name] + if active == nil { + active = make(map[int64]struct{}) + stat.active[user.Country.Name] = active + } + active[userID] = struct{}{} + } + retentionByOrigin, err := r.retentionStatsForDay(ctx, day, events[dayKey], users) + if err != nil { + return err + } + for _, statTimezone := range r.cfg.Dashboard.StatTimezones { + for _, sysOrigin := range r.cfg.Dashboard.SysOrigins { + if _, err := tx.ExecContext(ctx, ` +UPDATE country_dashboard_period_metric +SET daily_active_user = 0, + d1_retention_user = 0, d1_retention_base_user = 0, + d7_retention_user = 0, d7_retention_base_user = 0, + d30_retention_user = 0, d30_retention_base_user = 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 := r.upsertActiveRetentionRows(ctx, tx, day, statTimezone, sysOrigin, + statsByOrigin[sysOrigin], retentionByOrigin[sysOrigin]); err != nil { + return err + } + } + } + return nil +} + +func (r *MySQLRepository) retentionStatsForDay(ctx context.Context, day time.Time, + activeUsers map[int64]struct{}, users map[int64]dashboard.UserCountry) (map[string]map[string]activeRetentionValues, error) { + out := make(map[string]map[string]activeRetentionValues) + for _, days := range []int{1, 7, 30} { + bases, err := r.loadRetentionBase(ctx, day.AddDate(0, 0, -days)) + if err != nil { + return nil, err + } + for origin, countries := range bases { + originOut := out[origin] + if originOut == nil { + originOut = make(map[string]activeRetentionValues) + out[origin] = originOut + } + for country, item := range countries { + stats := originOut[country] + stats.countryName = item.name + switch days { + case 1: + stats.d1Base = item.count + case 7: + stats.d7Base = item.count + case 30: + stats.d30Base = item.count + } + originOut[country] = stats + } + } + } + for userID := range activeUsers { + user, ok := users[userID] + if !ok || user.CreatedAt.IsZero() || user.IsDeleted || !r.allowedOrigin(user.SysOrigin) { + continue + } + for _, days := range []int{1, 7, 30} { + if user.CreatedAt.Format("2006-01-02") != day.AddDate(0, 0, -days).Format("2006-01-02") { + continue + } + originOut := out[user.SysOrigin] + if originOut == nil { + originOut = make(map[string]activeRetentionValues) + out[user.SysOrigin] = originOut + } + country := user.Country.Code + stats := originOut[country] + if stats.countryName == "" { + stats.countryName = user.Country.Name + } + switch days { + case 1: + stats.d1User++ + case 7: + stats.d7User++ + case 30: + stats.d30User++ + } + originOut[country] = stats + } + } + return out, nil +} + +func (r *MySQLRepository) loadRetentionBase(ctx context.Context, cohortDay time.Time) (map[string]map[string]countryCount, error) { + args := stringArgs(r.cfg.Dashboard.SysOrigins) + args = append(args, cohortDay, cohortDay.AddDate(0, 0, 1)) + rows, err := r.db.QueryContext(ctx, ` +SELECT sys_origin, country_code, MAX(country_name), COUNT(1) +FROM dashboard_cdc_user_country +WHERE is_deleted = 0 + AND sys_origin IN `+inStringPlaceholders(len(r.cfg.Dashboard.SysOrigins))+` + AND user_created_at >= ? + AND user_created_at < ? +GROUP BY sys_origin, country_code`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + result := make(map[string]map[string]countryCount) + for rows.Next() { + var origin, country, name string + var count int64 + if err := rows.Scan(&origin, &country, &name, &count); err != nil { + return nil, err + } + countries := result[origin] + if countries == nil { + countries = make(map[string]countryCount) + result[origin] = countries + } + countries[country] = countryCount{name: name, count: count} + } + return result, rows.Err() +} + +func (r *MySQLRepository) upsertActiveRetentionRows(ctx context.Context, tx *sql.Tx, day time.Time, + statTimezone string, sysOrigin string, active map[string]*activeDayStats, retention map[string]activeRetentionValues) error { + values := make(map[string]activeRetentionValues) + for country, stat := range active { + item := values[country] + for name, users := range stat.active { + item.countryName = name + item.activeUsers += int64(len(users)) + } + values[country] = item + } + for country, item := range retention { + existing := values[country] + if existing.countryName == "" { + existing.countryName = item.countryName + } + existing.d1User = item.d1User + existing.d1Base = item.d1Base + existing.d7User = item.d7User + existing.d7Base = item.d7Base + existing.d30User = item.d30User + existing.d30Base = item.d30Base + values[country] = existing + } + if len(values) == 0 { + return nil + } + dayKey := day.Format("2006-01-02") + args := make([]interface{}, 0, len(values)*16) + for country, item := range values { + name := item.countryName + if name == "" { + name = country + } + args = append(args, dashboard.PeriodDay, dayKey, statTimezone, dayKey, day, + day.AddDate(0, 0, 1), sysOrigin, country, name, item.activeUsers, + item.d1User, item.d1Base, item.d7User, item.d7Base, item.d30User, item.d30Base) + } + _, 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, daily_active_user, + d1_retention_user, d1_retention_base_user, + d7_retention_user, d7_retention_base_user, + d30_retention_user, d30_retention_base_user, + refreshed_at +) VALUES `+placeholders(len(values), 16)+` +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), + daily_active_user = VALUES(daily_active_user), + d1_retention_user = VALUES(d1_retention_user), + d1_retention_base_user = VALUES(d1_retention_base_user), + d7_retention_user = VALUES(d7_retention_user), + d7_retention_base_user = VALUES(d7_retention_base_user), + d30_retention_user = VALUES(d30_retention_user), + d30_retention_base_user = VALUES(d30_retention_base_user), + refreshed_at = NOW()`, args...) + return err +} + +func (r *MySQLRepository) rebuildAllPeriodFromDays(ctx context.Context, tx *sql.Tx, statTimezone string) error { + for _, sysOrigin := range r.cfg.Dashboard.SysOrigins { + if _, err := tx.ExecContext(ctx, ` +DELETE FROM country_dashboard_period_user_metric +WHERE period_type = 'ALL' AND period_key = 'ALL' AND stat_timezone = ? AND sys_origin = ?`, statTimezone, sysOrigin); err != nil { + return err + } + if _, 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 +) +SELECT DISTINCT 'ALL', 'ALL', ?, '全部', NULL, NULL, + sys_origin, country_code, country_name, user_id, metric_type +FROM country_dashboard_period_user_metric +WHERE period_type = 'DAY' AND stat_timezone = ? AND sys_origin = ?`, statTimezone, statTimezone, sysOrigin); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, ` +DELETE FROM country_dashboard_period_metric +WHERE period_type = 'ALL' AND period_key = 'ALL' AND stat_timezone = ? AND sys_origin = ?`, statTimezone, sysOrigin); err != nil { + return err + } + 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, + country_new_user, daily_active_user, new_user_recharge, official_recharge, mifapay_recharge, + google_recharge, dealer_recharge, salary_exchange, salary_transfer, gift_consume, + lucky_gift_total_flow, lucky_gift_user, lucky_gift_payout, lucky_gift_anchor_share, + game_total_flow, game_user, game_payout, + d1_retention_user, d1_retention_base_user, d7_retention_user, d7_retention_base_user, + d30_retention_user, d30_retention_base_user, refreshed_at +) +SELECT 'ALL', 'ALL', ?, '全部', NULL, NULL, + d.sys_origin, d.country_code, MAX(d.country_name), + IFNULL(MAX(u.country_new_user), 0), + IFNULL(SUM(d.daily_active_user), 0), + IFNULL(SUM(d.new_user_recharge), 0), + IFNULL(SUM(d.official_recharge), 0), + IFNULL(SUM(d.mifapay_recharge), 0), + IFNULL(SUM(d.google_recharge), 0), + IFNULL(SUM(d.dealer_recharge), 0), + IFNULL(SUM(d.salary_exchange), 0), + IFNULL(SUM(d.salary_transfer), 0), + IFNULL(SUM(d.gift_consume), 0), + IFNULL(SUM(d.lucky_gift_total_flow), 0), + IFNULL(MAX(u.lucky_gift_user), 0), + IFNULL(SUM(d.lucky_gift_payout), 0), + IFNULL(SUM(d.lucky_gift_anchor_share), 0), + IFNULL(SUM(d.game_total_flow), 0), + IFNULL(MAX(u.game_user), 0), + IFNULL(SUM(d.game_payout), 0), + IFNULL(SUM(d.d1_retention_user), 0), + IFNULL(SUM(d.d1_retention_base_user), 0), + IFNULL(SUM(d.d7_retention_user), 0), + IFNULL(SUM(d.d7_retention_base_user), 0), + IFNULL(SUM(d.d30_retention_user), 0), + IFNULL(SUM(d.d30_retention_base_user), 0), + NOW() +FROM country_dashboard_period_metric d +LEFT JOIN ( + SELECT country_code, + SUM(metric_type = 'COUNTRY_NEW_USER') AS country_new_user, + SUM(metric_type = 'LUCKY_GIFT_USER') AS lucky_gift_user, + SUM(metric_type = 'GAME_USER') AS game_user + FROM country_dashboard_period_user_metric + WHERE period_type = 'ALL' AND period_key = 'ALL' AND stat_timezone = ? AND sys_origin = ? + GROUP BY country_code +) u ON u.country_code = d.country_code +WHERE d.period_type = 'DAY' AND d.stat_timezone = ? AND d.sys_origin = ? +GROUP BY d.sys_origin, d.country_code`, statTimezone, statTimezone, sysOrigin, statTimezone, sysOrigin); err != nil { + return err + } + } + return nil +} + +func inPlaceholders(count int) string { + return "(" + placeholders(count, 1) + ")" +} + +func inStringPlaceholders(count int) string { + return "(" + placeholders(count, 1) + ")" +} + +func stringArgs(values []string) []interface{} { + args := make([]interface{}, 0, len(values)) + for _, value := range values { + args = append(args, value) + } + return args +} diff --git a/internal/storage/gift_reconcile.go b/internal/storage/gift_reconcile.go index 55d8cdc..4248113 100644 --- a/internal/storage/gift_reconcile.go +++ b/internal/storage/gift_reconcile.go @@ -54,11 +54,6 @@ func (r *MySQLRepository) BuildGiftReconcileEvents(ctx context.Context, since ti UserCreatedAt: row.user.CreatedAt, } switch { - case row.eventType == "LUCKY_GIFT" && row.recordType == 1: - contribution.LuckyGiftTotalFlow = row.amount - contribution.LuckyGiftUser = true - case row.eventType == "LUCKY_GIFT" && row.recordType == 0: - contribution.LuckyGiftPayout = row.amount case isGiftConsumeEvent(row.eventType) && row.recordType == 1: contribution.GiftConsume = row.amount default: @@ -186,10 +181,9 @@ 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 = 'LUCKY_GIFT' - OR (w.event_type LIKE 'GIVE_GIFT%%' AND w.event_type NOT LIKE '%%LUCKY_GIFT%%' AND w.type = 1) - ) + 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, diff --git a/internal/storage/schema.go b/internal/storage/schema.go index 3560ce1..5d3ca4c 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -34,7 +34,8 @@ func (r *MySQLRepository) EnsureSchema(ctx context.Context) error { user_updated_at datetime NULL, updated_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (user_id), - KEY idx_dashboard_cdc_user_country_origin_country (sys_origin, country_code) + KEY idx_dashboard_cdc_user_country_origin_country (sys_origin, country_code), + KEY idx_dashboard_cdc_user_country_origin_created (sys_origin, user_created_at, country_code) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, createDailyMetricTable, createDailyUserMetricTable, @@ -128,8 +129,12 @@ func (r *MySQLRepository) ensureMetricColumns(ctx context.Context) error { "ALTER TABLE country_dashboard_daily_metric ADD COLUMN salary_transfer decimal(20,2) NOT NULL DEFAULT 0 AFTER salary_exchange"); err != nil { return err } - return r.addColumnIfMissing(ctx, "country_dashboard_period_metric", "salary_transfer", - "ALTER TABLE country_dashboard_period_metric ADD COLUMN salary_transfer decimal(20,2) NOT NULL DEFAULT 0 AFTER salary_exchange") + if err := r.addColumnIfMissing(ctx, "country_dashboard_period_metric", "salary_transfer", + "ALTER TABLE country_dashboard_period_metric ADD COLUMN salary_transfer decimal(20,2) NOT NULL DEFAULT 0 AFTER salary_exchange"); err != nil { + return err + } + return r.addIndexIfMissing(ctx, "dashboard_cdc_user_country", "idx_dashboard_cdc_user_country_origin_created", + "ALTER TABLE dashboard_cdc_user_country ADD INDEX idx_dashboard_cdc_user_country_origin_created (sys_origin, user_created_at, country_code)") } func (r *MySQLRepository) addColumnIfMissing(ctx context.Context, table string, column string, ddl string) error { @@ -153,6 +158,27 @@ WHERE table_schema = DATABASE() return nil } +func (r *MySQLRepository) addIndexIfMissing(ctx context.Context, table string, indexName string, ddl string) error { + var exists int + err := r.db.QueryRowContext(ctx, ` +SELECT COUNT(1) +FROM information_schema.statistics +WHERE table_schema = DATABASE() + AND table_name = ? + AND index_name = ? +`, table, indexName).Scan(&exists) + if err != nil { + return err + } + if exists > 0 { + return nil + } + if _, err := r.db.ExecContext(ctx, ddl); err != nil && !isDuplicateKeyError(err) { + return err + } + return nil +} + func isDuplicateColumnError(err error) bool { var mysqlErr *driverMysql.MySQLError if err == nil { @@ -164,6 +190,17 @@ func isDuplicateColumnError(err error) bool { return fmt.Sprint(err) == "Error 1060 (42S21): Duplicate column name 'salary_transfer'" } +func isDuplicateKeyError(err error) bool { + var mysqlErr *driverMysql.MySQLError + if err == nil { + return false + } + if errors.As(err, &mysqlErr) { + return mysqlErr.Number == 1061 + } + return false +} + const createPeriodUserMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_period_user_metric ( id bigint NOT NULL AUTO_INCREMENT, period_type varchar(16) NOT NULL,