From d39840a6edd7ca1deac2041be9d81a2eb01d32d8 Mon Sep 17 00:00:00 2001 From: ZuoZuo <68836346+Mrz-sakura@users.noreply.github.com> Date: Sun, 10 May 2026 02:05:01 +0800 Subject: [PATCH] Add task center MQ consumer and COS archive outbox --- cmd/api/main.go | 6 + cmd/consumer/main.go | 8 + go.mod | 5 + go.sum | 16 + internal/config/config.go | 53 ++++ internal/config/config_test.go | 40 +++ internal/model/taskcenter_models.go | 37 +++ internal/service/taskcenter/archive.go | 273 ++++++++++++++++++ internal/service/taskcenter/event.go | 194 ++++++++++--- internal/service/taskcenter/lifecycle.go | 202 +++++++++++++ .../service/taskcenter/taskcenter_test.go | 135 ++++++++- internal/service/taskcenter/types.go | 19 +- .../030_task_center_event_dedup_archive.sql | 35 +++ 13 files changed, 981 insertions(+), 42 deletions(-) create mode 100644 internal/service/taskcenter/archive.go create mode 100644 internal/service/taskcenter/lifecycle.go create mode 100644 migrations/030_task_center_event_dedup_archive.sql diff --git a/cmd/api/main.go b/cmd/api/main.go index be9746f..f77d68b 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -72,6 +72,12 @@ func main() { if err := weekStarService.StartMessageConsumer(workerCtx); err != nil { log.Fatalf("start week star message consumer failed: %v", err) } + if err := taskCenterService.StartMessageConsumer(workerCtx); err != nil { + log.Fatalf("start task center message consumer failed: %v", err) + } + if err := taskCenterService.StartArchiveWorker(workerCtx); err != nil { + log.Fatalf("start task center archive worker failed: %v", err) + } voiceRoomRedPacketService.StartRefundWorker(workerCtx) gameProviders := gameprovider.NewRegistry( diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index 3394ce0..2ced824 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -4,6 +4,7 @@ import ( "chatapp3-golang/internal/bootstrap" "chatapp3-golang/internal/service/regionimgroup" "chatapp3-golang/internal/service/registerreward" + "chatapp3-golang/internal/service/taskcenter" "chatapp3-golang/internal/service/voiceroomredpacket" "chatapp3-golang/internal/service/weekstar" "context" @@ -24,6 +25,7 @@ func main() { registerRewardService := registerreward.NewRegisterRewardService(app.Config, app.Repository.DB, app.Repository.Redis, &app.Gateways) weekStarService := weekstar.NewWeekStarService(app.Config, app.Repository.DB, app.Repository.Redis, &app.Gateways) + taskCenterService := taskcenter.NewService(app.Config, app.Repository.DB, &app.Gateways) regionIMGroupService := regionimgroup.NewService(app.Config, app.Repository.DB, &app.Gateways) voiceRoomRedPacketService := voiceroomredpacket.NewService(app.Config, app.Repository.DB, app.Repository.Redis, app.Gateways.Wallet, regionIMGroupService, app.Gateways.Room) voiceRoomRedPacketService.SetRegionResolver(&app.Gateways) @@ -37,6 +39,12 @@ func main() { if err := weekStarService.StartMessageConsumer(workerCtx); err != nil { log.Fatalf("start week star message consumer failed: %v", err) } + if err := taskCenterService.StartMessageConsumer(workerCtx); err != nil { + log.Fatalf("start task center message consumer failed: %v", err) + } + if err := taskCenterService.StartArchiveWorker(workerCtx); err != nil { + log.Fatalf("start task center archive worker failed: %v", err) + } voiceRoomRedPacketService.StartRefundWorker(workerCtx) log.Printf("chatapp3-golang consumer started") diff --git a/go.mod b/go.mod index 8161cc7..a6e3581 100644 --- a/go.mod +++ b/go.mod @@ -18,6 +18,7 @@ require ( github.com/bytedance/sonic/loader v0.1.1 // indirect github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/clbanning/mxj v1.8.4 // indirect github.com/cloudwego/base64x v0.1.4 // indirect github.com/cloudwego/iasm v0.2.0 // indirect github.com/dchest/siphash v1.2.3 // indirect @@ -32,6 +33,7 @@ require ( github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 // indirect github.com/golang/mock v1.6.0 // indirect github.com/golang/protobuf v1.5.4 // indirect + github.com/google/go-querystring v1.0.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 // indirect github.com/jinzhu/inflection v1.0.0 // indirect @@ -42,12 +44,15 @@ require ( github.com/leodido/go-urn v1.4.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/mattn/go-sqlite3 v1.14.22 // indirect + github.com/mitchellh/mapstructure v1.4.3 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/mozillazg/go-httpheader v0.2.1 // indirect github.com/natefinch/lumberjack v2.0.0+incompatible // indirect github.com/pelletier/go-toml/v2 v2.2.2 // indirect github.com/pierrec/lz4 v2.6.1+incompatible // indirect github.com/rogpeppe/go-internal v1.14.1 // indirect + github.com/tencentyun/cos-go-sdk-v5 v0.7.73 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.12 // indirect github.com/valyala/fastrand v1.1.0 // indirect diff --git a/go.sum b/go.sum index 6f1a539..4a8b6e5 100644 --- a/go.sum +++ b/go.sum @@ -49,6 +49,8 @@ github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI= github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI= github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU= +github.com/clbanning/mxj v1.8.4 h1:HuhwZtbyvyOw+3Z1AowPkU87JkJUSv751ELWaiTpj8I= +github.com/clbanning/mxj v1.8.4/go.mod h1:BVjHeAH+rl9rs6f+QIpeRl0tfu10SXn1pUSa5PVGJng= github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/cloudwego/base64x v0.1.4 h1:jwCgWpFanWmN8xoIUHa2rtzmkd5J2plF/dnLS6Xd/0Y= github.com/cloudwego/base64x v0.1.4/go.mod h1:0zlkT4Wn5C6NdauXdJRhSKRlJvmclQ1hhJgA0rcu/8w= @@ -94,6 +96,7 @@ github.com/go-sql-driver/mysql v1.7.0 h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ github.com/go-sql-driver/mysql v1.7.0/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI= 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-jwt/jwt/v5 v5.2.3/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= @@ -132,6 +135,8 @@ github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/ github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/go-querystring v1.0.0 h1:Xkwi/a1rcvNg1PPYe5vI8GbeBY/jrVuDX5ASuANWTrk= +github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs= github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc= @@ -140,6 +145,7 @@ github.com/google/pprof v0.0.0-20191218002539-d4f498aebedc/go.mod h1:ZgVRPoUq/hf github.com/google/pprof v0.0.0-20200212024743-f11f1df84d12/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM= github.com/google/pprof v0.0.0-20200229191704-1ebb73c60ed3/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM= github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI= +github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= @@ -179,11 +185,15 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU= github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= +github.com/mitchellh/mapstructure v1.4.3 h1:OVowDSCllw/YjdLkam3/sm7wEtOy59d8ndGgCcyj8cs= +github.com/mitchellh/mapstructure v1.4.3/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/mozillazg/go-httpheader v0.2.1 h1:geV7TrjbL8KXSyvghnFm+NyTux/hxwueTSrwhe88TQQ= +github.com/mozillazg/go-httpheader v0.2.1/go.mod h1:jJ8xECTlalr6ValeXYdOF8fFUISeBAdw6E61aqQma60= github.com/natefinch/lumberjack v2.0.0+incompatible h1:4QJd3OLAMgj7ph+yZTuX13Ld4UpgHp07nNdFX7mqFfM= github.com/natefinch/lumberjack v2.0.0+incompatible/go.mod h1:Wi9p2TTF5DG5oU+6YfsmYQpsTIOm0B1VNzQg9Mw6nPk= github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM= @@ -201,6 +211,7 @@ github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6L github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/rs/dnscache v0.0.0-20230804202142-fc85eb664529/go.mod h1:qe5TWALJ8/a1Lqznoc5BDHpYX/8HU60Hm2AwRmqzxqA= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= @@ -214,6 +225,11 @@ github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/common v1.0.563/go.mod h1:7sCQWVkxcsR38nffDW057DRGk8mUjK1Ing/EFOK8s8Y= +github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/kms v1.0.563/go.mod h1:uom4Nvi9W+Qkom0exYiJ9VWJjXwyxtPYTkKkaLMlfE0= +github.com/tencentyun/cos-go-sdk-v5 v0.7.73 h1:uFfgp1A7cQaAGR6QP9DsIkoEQ67b8ewj5r1RV6XB540= +github.com/tencentyun/cos-go-sdk-v5 v0.7.73/go.mod h1:STbTNaNKq03u+gscPEGOahKzLcGSYOj6Dzc5zNay7Pg= +github.com/tencentyun/qcloud-cos-sts-sdk v0.0.0-20250515025012-e0eec8a5d123/go.mod h1:b18KQa4IxHbxeseW1GcZox53d7J0z39VNONTxvvlkXw= github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE= diff --git a/internal/config/config.go b/internal/config/config.go index 95fc1ff..dd0fa07 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -88,6 +88,34 @@ type TaskCenterConfig struct { DefaultSysOrigin string Timezone string InternalEventTTL time.Duration + MQ TaskCenterMQConfig + Archive TaskCenterArchiveConfig +} + +// TaskCenterMQConfig 保存任务中心事件 MQ 消费配置。 +type TaskCenterMQConfig struct { + Enabled bool + Endpoint string + Namespace string + AccessKey string + AccessSecret string + SecurityToken string + ConsumerGroup string + Topic string + Tag string +} + +// TaskCenterArchiveConfig 保存任务中心原始事件 COS 归档配置。 +type TaskCenterArchiveConfig struct { + Enabled bool + Bucket string + Region string + SecretID string + SecretKey string + Prefix string + BatchSize int + FlushSeconds int + Workers int } // WeekStarConfig 保存周榜模块配置。 @@ -193,6 +221,9 @@ func Load() Config { weekStarRocketMQEnabledDefault := strings.TrimSpace(weekStarRocketMQEndpoint) != "" && strings.TrimSpace(weekStarRocketMQAccessKey) != "" && strings.TrimSpace(weekStarRocketMQAccessSecret) != "" + taskCenterMQEndpoint := getEnvAny([]string{"CHATAPP_TASK_CENTER_MQ_ENDPOINT", "TASK_CENTER_EVENT_MQ_ENDPOINT", "LIKEI_ROCKETMQ_ENDPOINT"}, "") + taskCenterMQAccessKey := getEnvAny([]string{"CHATAPP_TASK_CENTER_MQ_ACCESS_KEY", "TASK_CENTER_EVENT_MQ_ACCESS_KEY", "LIKEI_ROCKETMQ_ACCESS_KEY"}, "") + taskCenterMQAccessSecret := getEnvAny([]string{"CHATAPP_TASK_CENTER_MQ_ACCESS_SECRET", "TASK_CENTER_EVENT_MQ_SECRET_KEY", "LIKEI_ROCKETMQ_SECRET_KEY"}, "") return Config{ HTTP: HTTPConfig{ @@ -255,6 +286,28 @@ func Load() Config { []string{"CHATAPP_TASK_CENTER_EVENT_TTL_HOURS", "TASK_CENTER_EVENT_TTL_HOURS"}, 72, )) * time.Hour, + MQ: TaskCenterMQConfig{ + Enabled: getEnvBoolAny([]string{"CHATAPP_TASK_CENTER_MQ_ENABLED", "TASK_CENTER_EVENT_MQ_ENABLED"}, false), + Endpoint: taskCenterMQEndpoint, + Namespace: getEnvAny([]string{"CHATAPP_TASK_CENTER_MQ_NAMESPACE", "TASK_CENTER_EVENT_MQ_NAMESPACE"}, ""), + AccessKey: taskCenterMQAccessKey, + AccessSecret: taskCenterMQAccessSecret, + SecurityToken: getEnvAny([]string{"CHATAPP_TASK_CENTER_MQ_SECURITY_TOKEN", "TASK_CENTER_EVENT_MQ_SECURITY_TOKEN"}, ""), + ConsumerGroup: getEnvAny([]string{"CHATAPP_TASK_CENTER_MQ_CONSUMER_GROUP", "TASK_CENTER_EVENT_MQ_CONSUMER_GROUP"}, "task-center-progress"), + Topic: getEnvAny([]string{"CHATAPP_TASK_CENTER_MQ_TOPIC", "TASK_CENTER_EVENT_MQ_TOPIC"}, "RC_DEFAULT_APP_ORDINARY"), + Tag: getEnvAny([]string{"CHATAPP_TASK_CENTER_MQ_TAG", "TASK_CENTER_EVENT_MQ_TAG"}, "task_center_event"), + }, + Archive: TaskCenterArchiveConfig{ + Enabled: getEnvBoolAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_ENABLED", "TASK_CENTER_ARCHIVE_ENABLED"}, false), + Bucket: getEnvAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_COS_BUCKET", "TASK_CENTER_ARCHIVE_COS_BUCKET", "LIKEI_TENCENT_COS_BUCKET"}, ""), + Region: getEnvAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_COS_REGION", "TASK_CENTER_ARCHIVE_COS_REGION", "LIKEI_TENCENT_COS_REGION"}, ""), + SecretID: getEnvAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_COS_SECRET_ID", "TASK_CENTER_ARCHIVE_COS_SECRET_ID", "LIKEI_TENCENT_COS_SECRET_ID"}, ""), + SecretKey: getEnvAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_COS_SECRET_KEY", "TASK_CENTER_ARCHIVE_COS_SECRET_KEY", "LIKEI_TENCENT_COS_SECRET_KEY"}, ""), + Prefix: getEnvAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_COS_PREFIX", "TASK_CENTER_ARCHIVE_COS_PREFIX"}, "task-center-events/v1"), + BatchSize: getEnvIntAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_BATCH_SIZE", "TASK_CENTER_ARCHIVE_BATCH_SIZE"}, 10000), + FlushSeconds: getEnvIntAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_FLUSH_SECONDS", "TASK_CENTER_ARCHIVE_FLUSH_SECONDS"}, 60), + Workers: getEnvIntAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_WORKERS", "TASK_CENTER_ARCHIVE_WORKERS"}, 2), + }, }, WeekStar: WeekStarConfig{ DefaultSysOrigin: strings.ToUpper(getEnvAny([]string{"CHATAPP_WEEK_STAR_DEFAULT_SYS_ORIGIN", "WEEK_STAR_DEFAULT_SYS_ORIGIN"}, "LIKEI")), diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 68f7871..a536ad5 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -81,6 +81,46 @@ func TestLoadWeekStarRocketMQExplicitDisableOverridesCredentials(t *testing.T) { } } +func TestLoadTaskCenterMQUsesSharedRocketMQCredentials(t *testing.T) { + t.Setenv("CHATAPP_TASK_CENTER_MQ_ENABLED", "true") + t.Setenv("LIKEI_ROCKETMQ_ENDPOINT", "rmq.example.com:8081") + t.Setenv("LIKEI_ROCKETMQ_ACCESS_KEY", "access-key") + t.Setenv("LIKEI_ROCKETMQ_SECRET_KEY", "access-secret") + + cfg := Load() + if !cfg.TaskCenter.MQ.Enabled { + t.Fatalf("expected task center mq enabled") + } + if got := cfg.TaskCenter.MQ.Endpoint; got != "rmq.example.com:8081" { + t.Fatalf("task center mq endpoint = %q", got) + } + if got := cfg.TaskCenter.MQ.Topic; got != "RC_DEFAULT_APP_ORDINARY" { + t.Fatalf("task center mq topic = %q", got) + } + if got := cfg.TaskCenter.MQ.Tag; got != "task_center_event" { + t.Fatalf("task center mq tag = %q", got) + } +} + +func TestLoadTaskCenterArchiveUsesSharedCOSCredentials(t *testing.T) { + t.Setenv("CHATAPP_TASK_CENTER_ARCHIVE_ENABLED", "true") + t.Setenv("LIKEI_TENCENT_COS_BUCKET", "bucket-123") + t.Setenv("LIKEI_TENCENT_COS_REGION", "ap-test") + t.Setenv("LIKEI_TENCENT_COS_SECRET_ID", "secret-id") + t.Setenv("LIKEI_TENCENT_COS_SECRET_KEY", "secret-key") + + cfg := Load() + if !cfg.TaskCenter.Archive.Enabled { + t.Fatalf("expected task center archive enabled") + } + if got := cfg.TaskCenter.Archive.Bucket; got != "bucket-123" { + t.Fatalf("task center archive bucket = %q", got) + } + if got := cfg.TaskCenter.Archive.Prefix; got != "task-center-events/v1" { + t.Fatalf("task center archive prefix = %q", got) + } +} + func TestLoadLingxianConfigSupportsLikeiHkysSignKey(t *testing.T) { t.Setenv("LIKEI_GAME_HKYS_SIGN_KEY", "likei-hkys-key") diff --git a/internal/model/taskcenter_models.go b/internal/model/taskcenter_models.go index f009364..83c78f4 100644 --- a/internal/model/taskcenter_models.go +++ b/internal/model/taskcenter_models.go @@ -99,6 +99,43 @@ type TaskCenterEvent struct { func (TaskCenterEvent) TableName() string { return "task_center_event" } +// TaskCenterEventDedup 保存业务事件幂等记录。 +type TaskCenterEventDedup struct { + ID int64 `gorm:"column:id;primaryKey"` + SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_task_center_event_dedup,priority:1"` + EventID string `gorm:"column:event_id;size:128;uniqueIndex:uk_task_center_event_dedup,priority:2"` + EventType string `gorm:"column:event_type;size:64"` + UserID int64 `gorm:"column:user_id"` + Status string `gorm:"column:status;size:32;index:idx_task_center_event_dedup_status"` + FailureReason string `gorm:"column:failure_reason;size:512"` + CreateTime time.Time `gorm:"column:create_time;index:idx_task_center_event_dedup_create_time"` + UpdateTime time.Time `gorm:"column:update_time"` +} + +func (TaskCenterEventDedup) TableName() string { return "task_center_event_dedup" } + +// TaskCenterEventArchiveOutbox 保存待归档到 COS 的原始事件。 +type TaskCenterEventArchiveOutbox struct { + ID int64 `gorm:"column:id;primaryKey"` + SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_task_center_archive_event,priority:1"` + EventID string `gorm:"column:event_id;size:128;uniqueIndex:uk_task_center_archive_event,priority:2"` + EventType string `gorm:"column:event_type;size:64"` + UserID int64 `gorm:"column:user_id"` + OccurredAt time.Time `gorm:"column:occurred_at"` + RawJSON string `gorm:"column:raw_json;type:text"` + Status string `gorm:"column:status;size:32;index:idx_task_center_archive_status_time,priority:1"` + RetryCount int `gorm:"column:retry_count"` + COSKey string `gorm:"column:cos_key;size:512"` + FailureReason string `gorm:"column:failure_reason;size:512"` + CreateTime time.Time `gorm:"column:create_time;index:idx_task_center_archive_status_time,priority:2"` + UpdateTime time.Time `gorm:"column:update_time"` + ArchivedAt *time.Time `gorm:"column:archived_at"` +} + +func (TaskCenterEventArchiveOutbox) TableName() string { + return "task_center_event_archive_outbox" +} + // TaskCenterClaimRecord 保存任务奖励领取记录。 type TaskCenterClaimRecord struct { ID int64 `gorm:"column:id;primaryKey"` diff --git a/internal/service/taskcenter/archive.go b/internal/service/taskcenter/archive.go new file mode 100644 index 0000000..4c82603 --- /dev/null +++ b/internal/service/taskcenter/archive.go @@ -0,0 +1,273 @@ +package taskcenter + +import ( + "bytes" + "compress/gzip" + "context" + "fmt" + "log" + "net/http" + "net/url" + "path" + "sort" + "strings" + "time" + + "chatapp3-golang/internal/model" + + "github.com/tencentyun/cos-go-sdk-v5" + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +// StartArchiveWorker 启动任务中心事件 COS 归档 worker。 +func (s *Service) StartArchiveWorker(ctx context.Context) error { + if !s.cfg.TaskCenter.Archive.Enabled { + log.Printf("task center archive worker disabled") + return nil + } + if strings.TrimSpace(s.cfg.TaskCenter.Archive.Bucket) == "" || + strings.TrimSpace(s.cfg.TaskCenter.Archive.Region) == "" || + strings.TrimSpace(s.cfg.TaskCenter.Archive.SecretID) == "" || + strings.TrimSpace(s.cfg.TaskCenter.Archive.SecretKey) == "" { + log.Printf("task center archive worker skipped: cos bucket, region or credentials missing") + return nil + } + + workers := s.cfg.TaskCenter.Archive.Workers + if workers <= 0 { + workers = 1 + } + for index := 0; index < workers; index++ { + go s.runArchiveWorker(ctx, index) + } + return nil +} + +func (s *Service) runArchiveWorker(ctx context.Context, workerIndex int) { + interval := time.Duration(s.cfg.TaskCenter.Archive.FlushSeconds) * time.Second + if interval <= 0 { + interval = time.Minute + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + if ctx.Err() != nil { + return + } + if err := s.archivePendingEvents(ctx); err != nil { + log.Printf("task center archive worker %d failed: %v", workerIndex, err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +func (s *Service) archivePendingEvents(ctx context.Context) error { + batchSize := s.cfg.TaskCenter.Archive.BatchSize + if batchSize <= 0 { + batchSize = 10000 + } + rows, err := s.claimArchiveRows(ctx, batchSize) + if err != nil { + return err + } + if len(rows) == 0 { + return nil + } + + client, err := s.newCOSClient() + if err != nil { + return err + } + for _, group := range groupArchiveRows(rows) { + if err := s.uploadArchiveGroup(ctx, client, group); err != nil { + _ = s.markArchiveGroupFailed(ctx, group, err) + return err + } + } + return nil +} + +func (s *Service) claimArchiveRows(ctx context.Context, batchSize int) ([]model.TaskCenterEventArchiveOutbox, error) { + var rows []model.TaskCenterEventArchiveOutbox + err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + staleProcessingBefore := time.Now().Add(-10 * time.Minute) + if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}). + Where("status IN ? OR (status = ? AND update_time < ?)", + []string{ArchiveStatusPending, ArchiveStatusFailed}, + ArchiveStatusProcessing, + staleProcessingBefore, + ). + Order("create_time asc, id asc"). + Limit(batchSize). + Find(&rows).Error; err != nil { + return err + } + if len(rows) == 0 { + return nil + } + return tx.Model(&model.TaskCenterEventArchiveOutbox{}). + Where("id IN ?", archiveRowIDs(rows)). + Updates(map[string]any{ + "status": ArchiveStatusProcessing, + "update_time": time.Now(), + }).Error + }) + if err != nil { + return nil, err + } + return rows, nil +} + +func (s *Service) newCOSClient() (*cos.Client, error) { + bucketURL, err := url.Parse(fmt.Sprintf("https://%s.cos.%s.myqcloud.com", + strings.TrimSpace(s.cfg.TaskCenter.Archive.Bucket), + strings.TrimSpace(s.cfg.TaskCenter.Archive.Region), + )) + if err != nil { + return nil, err + } + baseURL := &cos.BaseURL{BucketURL: bucketURL} + return cos.NewClient(baseURL, &http.Client{ + Transport: &cos.AuthorizationTransport{ + SecretID: strings.TrimSpace(s.cfg.TaskCenter.Archive.SecretID), + SecretKey: strings.TrimSpace(s.cfg.TaskCenter.Archive.SecretKey), + }, + }), nil +} + +func groupArchiveRows(rows []model.TaskCenterEventArchiveOutbox) [][]model.TaskCenterEventArchiveOutbox { + groupsByKey := make(map[string][]model.TaskCenterEventArchiveOutbox) + keys := make([]string, 0) + for _, row := range rows { + groupKey := row.SysOrigin + "\x00" + row.EventType + if _, exists := groupsByKey[groupKey]; !exists { + keys = append(keys, groupKey) + } + groupsByKey[groupKey] = append(groupsByKey[groupKey], row) + } + sort.Strings(keys) + groups := make([][]model.TaskCenterEventArchiveOutbox, 0, len(keys)) + for _, key := range keys { + groups = append(groups, groupsByKey[key]) + } + return groups +} + +func (s *Service) uploadArchiveGroup(ctx context.Context, client *cos.Client, rows []model.TaskCenterEventArchiveOutbox) error { + if len(rows) == 0 { + return nil + } + cosKey := s.buildArchiveCOSKey(rows) + body, err := gzipArchiveRows(rows) + if err != nil { + return err + } + _, err = client.Object.Put(ctx, cosKey, bytes.NewReader(body), &cos.ObjectPutOptions{ + ObjectPutHeaderOptions: &cos.ObjectPutHeaderOptions{ + ContentType: "application/gzip", + ContentEncoding: "gzip", + }, + }) + if err != nil { + return err + } + return s.markArchiveGroupUploaded(ctx, rows, cosKey) +} + +func gzipArchiveRows(rows []model.TaskCenterEventArchiveOutbox) ([]byte, error) { + var buffer bytes.Buffer + writer := gzip.NewWriter(&buffer) + for _, row := range rows { + line := strings.TrimSpace(row.RawJSON) + if line == "" { + line = "{}" + } + if _, err := writer.Write([]byte(line)); err != nil { + _ = writer.Close() + return nil, err + } + if _, err := writer.Write([]byte("\n")); err != nil { + _ = writer.Close() + return nil, err + } + } + if err := writer.Close(); err != nil { + return nil, err + } + return buffer.Bytes(), nil +} + +func (s *Service) buildArchiveCOSKey(rows []model.TaskCenterEventArchiveOutbox) string { + first := rows[0] + occurredAt := first.OccurredAt + if occurredAt.IsZero() { + occurredAt = time.Now() + } + occurredAt = occurredAt.UTC() + prefix := strings.Trim(strings.TrimSpace(s.cfg.TaskCenter.Archive.Prefix), "/") + if prefix == "" { + prefix = "task-center-events/v1" + } + fileName := fmt.Sprintf("part-%s-%d-%d.jsonl.gz", + time.Now().UTC().Format("20060102T150405.000000000Z"), + first.ID, + len(rows), + ) + return path.Join( + prefix, + "dt="+occurredAt.Format("2006-01-02"), + "hour="+occurredAt.Format("15"), + "sys_origin="+sanitizeArchivePartition(first.SysOrigin), + "event_type="+sanitizeArchivePartition(first.EventType), + fileName, + ) +} + +func sanitizeArchivePartition(value string) string { + value = strings.TrimSpace(value) + if value == "" { + return "UNKNOWN" + } + replacer := strings.NewReplacer("/", "_", "\\", "_", " ", "_", "=", "_") + return replacer.Replace(value) +} + +func (s *Service) markArchiveGroupUploaded(ctx context.Context, rows []model.TaskCenterEventArchiveOutbox, cosKey string) error { + ids := archiveRowIDs(rows) + now := time.Now() + return s.db.WithContext(ctx).Model(&model.TaskCenterEventArchiveOutbox{}). + Where("id IN ?", ids). + Updates(map[string]any{ + "status": ArchiveStatusUploaded, + "cos_key": cosKey, + "failure_reason": "", + "archived_at": now, + "update_time": now, + }).Error +} + +func (s *Service) markArchiveGroupFailed(ctx context.Context, rows []model.TaskCenterEventArchiveOutbox, archiveErr error) error { + ids := archiveRowIDs(rows) + return s.db.WithContext(ctx).Model(&model.TaskCenterEventArchiveOutbox{}). + Where("id IN ?", ids). + Updates(map[string]any{ + "status": ArchiveStatusFailed, + "retry_count": gorm.Expr("retry_count + 1"), + "failure_reason": truncateTaskCenterFailure(archiveErr.Error()), + "update_time": time.Now(), + }).Error +} + +func archiveRowIDs(rows []model.TaskCenterEventArchiveOutbox) []int64 { + ids := make([]int64, 0, len(rows)) + for _, row := range rows { + ids = append(ids, row.ID) + } + return ids +} diff --git a/internal/service/taskcenter/event.go b/internal/service/taskcenter/event.go index c734fb3..88fce65 100644 --- a/internal/service/taskcenter/event.go +++ b/internal/service/taskcenter/event.go @@ -13,6 +13,7 @@ import ( "chatapp3-golang/internal/utils" "gorm.io/gorm" + "gorm.io/gorm/clause" ) // ReportSimpleEvent 供 Go 内部业务模块上报任务进度事件。 @@ -23,7 +24,7 @@ func (s *Service) ReportSimpleEvent(ctx context.Context, sysOrigin, eventID, eve // ReportSimpleEventAt 和 ReportSimpleEvent 相同，但允许 MQ 等异步事件传入真实发生时间。 func (s *Service) ReportSimpleEventAt(ctx context.Context, sysOrigin, eventID, eventType string, userID, deltaValue int64, occurredAt time.Time, payload map[string]any) error { - _, err := s.ReceiveEvent(ctx, EventRequest{ + _, err := s.ProcessEvent(ctx, EventRequest{ SysOrigin: sysOrigin, EventID: eventID, EventType: eventType, @@ -40,6 +41,11 @@ func (s *Service) ReportSimpleEventAt(ctx context.Context, sysOrigin, eventID, e // ReceiveEvent 接收后端业务事件并更新命中的任务进度。 func (s *Service) ReceiveEvent(ctx context.Context, req EventRequest) (*EventResponse, error) { + return s.ProcessEvent(ctx, req) +} + +// ProcessEvent 统一处理 HTTP 和 MQ 上报的任务事件。 +func (s *Service) ProcessEvent(ctx context.Context, req EventRequest) (*EventResponse, error) { sysOrigin := s.normalizeSysOrigin(req.SysOrigin) eventID := strings.TrimSpace(req.EventID) if eventID == "" { @@ -59,55 +65,62 @@ func (s *Service) ReceiveEvent(ctx context.Context, req EventRequest) (*EventRes if req.DeltaValue <= 0 { return nil, NewAppError(http.StatusBadRequest, "invalid_delta_value", "deltaValue must be greater than 0") } - - var existingCount int64 - if err := s.db.WithContext(ctx).Model(&model.TaskCenterEvent{}). - Where("sys_origin = ? AND event_id = ?", sysOrigin, eventID). - Count(&existingCount).Error; err != nil { - return nil, err - } - if existingCount > 0 { - return &EventResponse{Processed: false, Updated: 0, Reason: "duplicate_event"}, nil - } + occurredAt := parseOccurredAt(req.OccurredAt, time.Now()) + receivedAt := time.Now() + rawJSON := buildArchiveRawJSON(req, sysOrigin, eventID, eventType, userID, req.DeltaValue, occurredAt, receivedAt, "", "") + payloadJSON := marshalPayloadJSON(req.Payload) bundle, err := s.loadEnabledBundle(ctx, sysOrigin) if err != nil { return nil, err } - if bundle == nil { - return &EventResponse{Processed: false, Updated: 0, Reason: "task_center_disabled"}, nil - } matchedTasks := make([]model.TaskCenterTaskConfig, 0) - for _, task := range bundle.Tasks { - if strings.EqualFold(task.ConditionType, eventType) { - matchedTasks = append(matchedTasks, task) + if bundle != nil { + for _, task := range bundle.Tasks { + if strings.EqualFold(task.ConditionType, eventType) { + matchedTasks = append(matchedTasks, task) + } } } - if len(matchedTasks) == 0 { - return &EventResponse{Processed: false, Updated: 0, Reason: "no_matched_task"}, nil - } - occurredAt := parseOccurredAt(req.OccurredAt, time.Now()) - rawJSON := "" - if len(req.Payload) > 0 { - if bytes, marshalErr := json.Marshal(req.Payload); marshalErr == nil { - rawJSON = string(bytes) - } + resultReason := "" + if bundle == nil { + resultReason = "task_center_disabled" + } else if len(matchedTasks) == 0 { + resultReason = "no_matched_task" } updated := 0 + duplicate := false if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { - var duplicateCount int64 - if err := tx.Model(&model.TaskCenterEvent{}). - Where("sys_origin = ? AND event_id = ?", sysOrigin, eventID). - Count(&duplicateCount).Error; err != nil { + inserted, err := insertTaskEventDedupTx(tx, sysOrigin, eventID, eventType, userID) + if err != nil { return err } - if duplicateCount > 0 { + if !inserted { + duplicate = true return nil } + var legacyEventCount int64 + if err := tx.Model(&model.TaskCenterEvent{}). + Where("sys_origin = ? AND event_id = ?", sysOrigin, eventID). + Count(&legacyEventCount).Error; err != nil { + return err + } + if legacyEventCount > 0 { + duplicate = true + return updateTaskEventDedupTx(tx, sysOrigin, eventID, DedupStatusIgnored, "duplicate_event") + } + + if err := createTaskEventArchiveOutboxTx(tx, sysOrigin, eventID, eventType, userID, occurredAt, rawJSON); err != nil { + return err + } + if resultReason != "" { + return updateTaskEventDedupTx(tx, sysOrigin, eventID, DedupStatusIgnored, resultReason) + } + for _, task := range matchedTasks { cycleKey := currentCycleKey(task.TaskCategory, bundle.Location, occurredAt) progress, err := s.loadOrCreateProgressTx(tx, sysOrigin, userID, task, cycleKey) @@ -145,7 +158,7 @@ func (s *Service) ReceiveEvent(ctx context.Context, req EventRequest) (*EventRes CycleKey: cycleKey, DeltaValue: req.DeltaValue, CompletedValue: progress.CompletedValue, - RawJSON: rawJSON, + RawJSON: payloadJSON, OccurredAt: occurredAt, CreateTime: time.Now(), UpdateTime: time.Now(), @@ -155,16 +168,123 @@ func (s *Service) ReceiveEvent(ctx context.Context, req EventRequest) (*EventRes } updated++ } - return nil + if updated == 0 { + resultReason = "no_updated_task" + return updateTaskEventDedupTx(tx, sysOrigin, eventID, DedupStatusIgnored, resultReason) + } + return updateTaskEventDedupTx(tx, sysOrigin, eventID, DedupStatusSuccess, "") }); err != nil { return nil, err } - reason := "" - if updated == 0 { - reason = "no_updated_task" + if duplicate { + return &EventResponse{Processed: false, Updated: 0, Reason: "duplicate_event"}, nil } - return &EventResponse{Processed: updated > 0, Updated: updated, Reason: reason}, nil + return &EventResponse{Processed: updated > 0, Updated: updated, Reason: resultReason}, nil +} + +func insertTaskEventDedupTx(tx *gorm.DB, sysOrigin, eventID, eventType string, userID int64) (bool, error) { + id, err := utils.NextID() + if err != nil { + return false, err + } + now := time.Now() + row := model.TaskCenterEventDedup{ + ID: id, + SysOrigin: sysOrigin, + EventID: eventID, + EventType: eventType, + UserID: userID, + Status: DedupStatusProcessing, + CreateTime: now, + UpdateTime: now, + } + result := tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "sys_origin"}, {Name: "event_id"}}, + DoNothing: true, + }).Create(&row) + if result.Error != nil { + return false, result.Error + } + return result.RowsAffected > 0, nil +} + +func updateTaskEventDedupTx(tx *gorm.DB, sysOrigin, eventID, status, failureReason string) error { + return tx.Model(&model.TaskCenterEventDedup{}). + Where("sys_origin = ? AND event_id = ?", sysOrigin, eventID). + Updates(map[string]any{ + "status": status, + "failure_reason": truncateTaskCenterFailure(failureReason), + "update_time": time.Now(), + }).Error +} + +func createTaskEventArchiveOutboxTx(tx *gorm.DB, sysOrigin, eventID, eventType string, userID int64, occurredAt time.Time, rawJSON string) error { + id, err := utils.NextID() + if err != nil { + return err + } + now := time.Now() + row := model.TaskCenterEventArchiveOutbox{ + ID: id, + SysOrigin: sysOrigin, + EventID: eventID, + EventType: eventType, + UserID: userID, + OccurredAt: occurredAt, + RawJSON: rawJSON, + Status: ArchiveStatusPending, + CreateTime: now, + UpdateTime: now, + } + return tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "sys_origin"}, {Name: "event_id"}}, + DoNothing: true, + }).Create(&row).Error +} + +func marshalPayloadJSON(payload map[string]any) string { + if len(payload) == 0 { + return "" + } + bytes, err := json.Marshal(payload) + if err != nil { + return "" + } + return string(bytes) +} + +func buildArchiveRawJSON(req EventRequest, sysOrigin, eventID, eventType string, userID, deltaValue int64, occurredAt, receivedAt time.Time, processResult, processReason string) string { + row := map[string]any{ + "schema_version": 1, + "sys_origin": sysOrigin, + "event_id": eventID, + "event_type": eventType, + "user_id": userID, + "delta_value": deltaValue, + "occurred_at": occurredAt.UTC().Format(time.RFC3339Nano), + "received_at": receivedAt.UTC().Format(time.RFC3339Nano), + "payload": req.Payload, + } + if processResult != "" { + row["process_result"] = processResult + } + if processReason != "" { + row["process_reason"] = processReason + } + bytes, err := json.Marshal(row) + if err != nil { + return "{}" + } + return string(bytes) +} + +func truncateTaskCenterFailure(value string) string { + value = strings.TrimSpace(value) + if len(value) <= 512 { + return value + } + return value[:512] } func (s *Service) loadOrCreateProgressTx(tx *gorm.DB, sysOrigin string, userID int64, task model.TaskCenterTaskConfig, cycleKey string) (model.TaskCenterUserProgress, error) { diff --git a/internal/service/taskcenter/lifecycle.go b/internal/service/taskcenter/lifecycle.go new file mode 100644 index 0000000..d82742f --- /dev/null +++ b/internal/service/taskcenter/lifecycle.go @@ -0,0 +1,202 @@ +package taskcenter + +import ( + "context" + "encoding/json" + "errors" + "log" + "net/http" + "strings" + "time" + + "chatapp3-golang/internal/common" + + rmq "github.com/apache/rocketmq-clients/golang/v5" + "github.com/apache/rocketmq-clients/golang/v5/credentials" +) + +// StartMessageConsumer 启动任务中心事件 MQ 消费链路。 +func (s *Service) StartMessageConsumer(ctx context.Context) error { + if !s.cfg.TaskCenter.MQ.Enabled { + log.Printf("task center rocketmq consumer disabled") + return nil + } + if strings.TrimSpace(s.cfg.TaskCenter.MQ.Endpoint) == "" || + strings.TrimSpace(s.cfg.TaskCenter.MQ.AccessKey) == "" || + strings.TrimSpace(s.cfg.TaskCenter.MQ.AccessSecret) == "" || + strings.TrimSpace(s.cfg.TaskCenter.MQ.Topic) == "" || + strings.TrimSpace(s.cfg.TaskCenter.MQ.ConsumerGroup) == "" { + log.Printf("task center rocketmq consumer skipped: endpoint, credentials, topic or consumer group missing") + return nil + } + + go s.startTaskCenterRocketMQConsumerWithRetry(ctx) + return nil +} + +func (s *Service) startTaskCenterRocketMQConsumerWithRetry(ctx context.Context) { + retryDelay := 5 * time.Second + for { + if ctx.Err() != nil { + return + } + if err := s.startTaskCenterRocketMQConsumer(ctx); err != nil { + log.Printf("start task center rocketmq consumer failed: %v; retry in %s", err, retryDelay) + timer := time.NewTimer(retryDelay) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + if retryDelay < time.Minute { + retryDelay *= 2 + if retryDelay > time.Minute { + retryDelay = time.Minute + } + } + continue + } + return + } +} + +func (s *Service) startTaskCenterRocketMQConsumer(ctx context.Context) error { + filter := rmq.SUB_ALL + if tag := strings.TrimSpace(s.cfg.TaskCenter.MQ.Tag); tag != "" && tag != "*" { + filter = rmq.NewFilterExpression(tag) + } + + consumer, err := rmq.NewPushConsumer(&rmq.Config{ + Endpoint: s.cfg.TaskCenter.MQ.Endpoint, + NameSpace: s.cfg.TaskCenter.MQ.Namespace, + ConsumerGroup: s.cfg.TaskCenter.MQ.ConsumerGroup, + Credentials: &credentials.SessionCredentials{ + AccessKey: s.cfg.TaskCenter.MQ.AccessKey, + AccessSecret: s.cfg.TaskCenter.MQ.AccessSecret, + SecurityToken: s.cfg.TaskCenter.MQ.SecurityToken, + }, + }, + rmq.WithPushSubscriptionExpressions(map[string]*rmq.FilterExpression{ + s.cfg.TaskCenter.MQ.Topic: filter, + }), + rmq.WithPushConsumptionThreadCount(8), + rmq.WithPushMaxCacheMessageCount(512), + rmq.WithPushMessageListener(&rmq.FuncMessageListener{ + Consume: func(messageView *rmq.MessageView) rmq.ConsumerResult { + if messageView == nil { + return rmq.SUCCESS + } + if err := s.processTaskCenterMQMessage(context.Background(), messageView); err != nil { + log.Printf("task center mq process failed. messageId=%s err=%v", messageView.GetMessageId(), err) + return rmq.FAILURE + } + return rmq.SUCCESS + }, + }), + ) + if err != nil { + return err + } + if err := consumer.Start(); err != nil { + return err + } + s.rocketConsumer = consumer + log.Printf("task center rocketmq consumer started. topic=%s group=%s tag=%s", + s.cfg.TaskCenter.MQ.Topic, + s.cfg.TaskCenter.MQ.ConsumerGroup, + s.cfg.TaskCenter.MQ.Tag, + ) + go func() { + <-ctx.Done() + if s.rocketConsumer != nil { + _ = s.rocketConsumer.GracefulStop() + } + }() + return nil +} + +func (s *Service) processTaskCenterMQMessage(ctx context.Context, messageView *rmq.MessageView) error { + payload := strings.TrimSpace(string(messageView.GetBody())) + if payload == "" { + return nil + } + req, err := decodeTaskCenterEventPayload(payload, s.cfg.TaskCenter.MQ.Tag) + if err != nil { + var syntaxErr *json.SyntaxError + var typeErr *json.UnmarshalTypeError + if errors.As(err, &syntaxErr) || errors.As(err, &typeErr) { + log.Printf("drop malformed task center message. messageId=%s err=%v", messageView.GetMessageId(), err) + return nil + } + return err + } + if strings.TrimSpace(req.EventID) == "" { + return nil + } + if _, err := s.ProcessEvent(ctx, req); err != nil { + var appErr *common.AppError + if errors.As(err, &appErr) && appErr.Status >= http.StatusBadRequest && appErr.Status < http.StatusInternalServerError { + log.Printf("drop invalid task center message. messageId=%s code=%s err=%v", messageView.GetMessageId(), appErr.Code, err) + return nil + } + return err + } + return nil +} + +func decodeTaskCenterEventPayload(payload string, expectedTag string) (EventRequest, error) { + var req EventRequest + if err := json.Unmarshal([]byte(payload), &req); err != nil { + return req, err + } + if strings.TrimSpace(req.EventID) != "" || strings.TrimSpace(req.EventType) != "" || req.UserID.Int64() > 0 { + return req, nil + } + + var envelope taskCenterMessageEventEnvelope + if err := json.Unmarshal([]byte(payload), &envelope); err != nil { + return req, err + } + if shouldSkipTaskCenterEnvelope(envelope.Tag, expectedTag) { + return EventRequest{}, nil + } + bodyPayload, ok, err := envelope.bodyPayload() + if err != nil || !ok { + return EventRequest{}, err + } + if err := json.Unmarshal([]byte(bodyPayload), &req); err != nil { + return EventRequest{}, err + } + return req, nil +} + +type taskCenterMessageEventEnvelope struct { + Tag string `json:"tag"` + Body json.RawMessage `json:"body"` +} + +func (e taskCenterMessageEventEnvelope) bodyPayload() (string, bool, error) { + raw := strings.TrimSpace(string(e.Body)) + if raw == "" || raw == "null" { + return "", false, nil + } + if strings.HasPrefix(raw, "\"") { + var body string + if err := json.Unmarshal(e.Body, &body); err != nil { + return "", false, err + } + body = strings.TrimSpace(body) + return body, body != "", nil + } + return raw, true, nil +} + +func shouldSkipTaskCenterEnvelope(actualTag, expectedTag string) bool { + actualTag = strings.TrimSpace(actualTag) + expectedTag = strings.TrimSpace(expectedTag) + if actualTag == "" || expectedTag == "" || expectedTag == "*" { + return false + } + return actualTag != expectedTag +} diff --git a/internal/service/taskcenter/taskcenter_test.go b/internal/service/taskcenter/taskcenter_test.go index 5be61fa..4ae87ae 100644 --- a/internal/service/taskcenter/taskcenter_test.go +++ b/internal/service/taskcenter/taskcenter_test.go @@ -2,7 +2,9 @@ package taskcenter import ( "context" + "strings" "testing" + "time" "chatapp3-golang/internal/common" "chatapp3-golang/internal/config" @@ -25,6 +27,8 @@ func newTestService(t *testing.T, java taskCenterJavaGateway) (*Service, *gorm.D &model.TaskCenterConditionJumpConfig{}, &model.TaskCenterUserProgress{}, &model.TaskCenterEvent{}, + &model.TaskCenterEventDedup{}, + &model.TaskCenterEventArchiveOutbox{}, &model.TaskCenterClaimRecord{}, ); err != nil { t.Fatalf("auto migrate: %v", err) @@ -40,7 +44,7 @@ func newTestService(t *testing.T, java taskCenterJavaGateway) (*Service, *gorm.D func TestDailyTaskConfigEventListAndClaim(t *testing.T) { java := &taskCenterTestGateway{events: map[string]bool{}} - service, _ := newTestService(t, java) + service, db := newTestService(t, java) ctx := context.Background() configResp, err := service.SaveConfig(ctx, SaveConfigRequest{ @@ -92,6 +96,37 @@ func TestDailyTaskConfigEventListAndClaim(t *testing.T) { if !eventResp.Processed || eventResp.Updated != 1 { t.Fatalf("ReceiveEvent() resp = %+v", eventResp) } + duplicateResp, err := service.ReceiveEvent(ctx, EventRequest{ + SysOrigin: "LIKEI", + EventID: "event-1", + EventType: "MIC_DURATION_SECONDS", + UserID: flexibleInt64(user.UserID), + DeltaValue: 60, + }) + if err != nil { + t.Fatalf("ReceiveEvent() duplicate error = %v", err) + } + if duplicateResp.Processed || duplicateResp.Reason != "duplicate_event" { + t.Fatalf("ReceiveEvent() duplicate resp = %+v", duplicateResp) + } + var dedupCount int64 + if err := db.Model(&model.TaskCenterEventDedup{}). + Where("sys_origin = ? AND event_id = ?", "LIKEI", "event-1"). + Count(&dedupCount).Error; err != nil { + t.Fatalf("count dedup: %v", err) + } + if dedupCount != 1 { + t.Fatalf("dedup rows = %d, want 1", dedupCount) + } + var outboxCount int64 + if err := db.Model(&model.TaskCenterEventArchiveOutbox{}). + Where("sys_origin = ? AND event_id = ?", "LIKEI", "event-1"). + Count(&outboxCount).Error; err != nil { + t.Fatalf("count archive outbox: %v", err) + } + if outboxCount != 1 { + t.Fatalf("archive outbox rows = %d, want 1", outboxCount) + } tasks, err = service.ListTasks(ctx, user) if err != nil { @@ -196,6 +231,104 @@ func TestExclusiveTaskConfigUsesLifetimeCycleAndHidesAfterClaim(t *testing.T) { } } +func TestProcessEventWithProductionGameEventShape(t *testing.T) { + service, db := newTestService(t, nil) + ctx := context.Background() + now := time.Now() + configRow := model.TaskCenterConfig{ + ID: 2052721146352111616, + SysOrigin: "LIKEI", + Enabled: true, + Timezone: "Asia/Riyadh", + CreateTime: now, + UpdateTime: now, + } + if err := db.Create(&configRow).Error; err != nil { + t.Fatalf("create config: %v", err) + } + tasks := []model.TaskCenterTaskConfig{ + { + ID: 2053002284639457280, + ConfigID: configRow.ID, + SysOrigin: "LIKEI", + TaskCode: "GAME_200K", + TaskCategory: TaskCategoryDaily, + Enabled: true, + TaskName: "Spend 200K coins in the game.", + TaskDesc: "progress", + ConditionType: EventTypeGameConsumeGold, + TargetValue: 200000, + TargetUnit: "gold", + RewardGold: 500, + JumpType: "ROOM_RANDOM_WITH_MIC", + JumpPage: "/room?gameId=0", + SortOrder: 30, + CreateTime: now, + UpdateTime: now, + }, + { + ID: 2053005491679793152, + ConfigID: configRow.ID, + SysOrigin: "LIKEI", + TaskCode: "GAME_40K", + TaskCategory: TaskCategoryNewbie, + Enabled: true, + TaskName: "Spend 40K coins in the game.", + TaskDesc: "progress", + ConditionType: EventTypeGameConsumeGold, + TargetValue: 40000, + TargetUnit: "gold", + RewardGold: 200, + JumpType: "ROOM_RANDOM_WITH_MIC", + SortOrder: 30, + CreateTime: now, + UpdateTime: now, + }, + } + if err := db.Create(&tasks).Error; err != nil { + t.Fatalf("create tasks: %v", err) + } + + resp, err := service.ProcessEvent(ctx, EventRequest{ + SysOrigin: "LIKEI", + EventID: "BAISHUN_GAME_CONSUME:LIKEI:ee2d4218-4f58-406a-bf0f-d6b09f7e9930", + EventType: EventTypeGameConsumeGold, + UserID: flexibleInt64(2053169065266049025), + DeltaValue: 100, + OccurredAt: "2026-05-09T20:48:30.291+03:00", + Payload: map[string]any{ + "diffMsg": "bet", + "gameId": float64(1067), + "gameRoundId": "20260510000171", + "orderId": "ee2d4218-4f58-406a-bf0f-d6b09f7e9930", + "provider": "BAISHUN", + }, + }) + if err != nil { + t.Fatalf("ProcessEvent() error = %v", err) + } + if !resp.Processed || resp.Updated != 2 { + t.Fatalf("ProcessEvent() resp = %+v, want processed with two task updates", resp) + } + var progressCount int64 + if err := db.Model(&model.TaskCenterUserProgress{}). + Where("sys_origin = ? AND user_id = ? AND completed_value = ?", "LIKEI", int64(2053169065266049025), int64(100)). + Count(&progressCount).Error; err != nil { + t.Fatalf("count progress: %v", err) + } + if progressCount != 2 { + t.Fatalf("progress rows = %d, want 2", progressCount) + } + var outbox model.TaskCenterEventArchiveOutbox + if err := db.Where("sys_origin = ? AND event_id = ?", "LIKEI", "BAISHUN_GAME_CONSUME:LIKEI:ee2d4218-4f58-406a-bf0f-d6b09f7e9930"). + First(&outbox).Error; err != nil { + t.Fatalf("load archive outbox: %v", err) + } + if outbox.Status != ArchiveStatusPending || !strings.Contains(outbox.RawJSON, "BAISHUN_GAME_CONSUME") { + t.Fatalf("archive outbox = %+v", outbox) + } +} + func TestConditionJumpConfigAppliesToTasks(t *testing.T) { service, _ := newTestService(t, nil) ctx := context.Background() diff --git a/internal/service/taskcenter/types.go b/internal/service/taskcenter/types.go index 5ef9f8b..299957b 100644 --- a/internal/service/taskcenter/types.go +++ b/internal/service/taskcenter/types.go @@ -13,6 +13,7 @@ import ( "chatapp3-golang/internal/config" "chatapp3-golang/internal/integration" + rmq "github.com/apache/rocketmq-clients/golang/v5" "gorm.io/gorm" ) @@ -32,6 +33,15 @@ const ( ClaimStatusSuccess = "SUCCESS" ClaimStatusFailed = "FAILED" + DedupStatusProcessing = "PROCESSING" + DedupStatusSuccess = "SUCCESS" + DedupStatusIgnored = "IGNORED" + + ArchiveStatusPending = "PENDING" + ArchiveStatusProcessing = "PROCESSING" + ArchiveStatusUploaded = "UPLOADED" + ArchiveStatusFailed = "FAILED" + JumpTypeRechargeInApp = "RECHARGE_IN_APP" defaultTaskCenterTimezone = "Asia/Riyadh" @@ -54,10 +64,11 @@ type taskCenterJavaGateway interface { // Service 负责任务配置、进度、领取和事件入库。 type Service struct { - cfg config.Config - db taskCenterDB - java taskCenterJavaGateway - location *time.Location + cfg config.Config + db taskCenterDB + java taskCenterJavaGateway + location *time.Location + rocketConsumer rmq.PushConsumer } func NewService(cfg config.Config, db taskCenterDB, java taskCenterJavaGateway) *Service { diff --git a/migrations/030_task_center_event_dedup_archive.sql b/migrations/030_task_center_event_dedup_archive.sql new file mode 100644 index 0000000..ab692a4 --- /dev/null +++ b/migrations/030_task_center_event_dedup_archive.sql @@ -0,0 +1,35 @@ +CREATE TABLE IF NOT EXISTS `task_center_event_dedup` ( + `id` bigint NOT NULL COMMENT '主键ID', + `sys_origin` varchar(32) NOT NULL COMMENT '来源系统', + `event_id` varchar(128) NOT NULL COMMENT '业务事件ID', + `event_type` varchar(64) NOT NULL COMMENT '事件类型', + `user_id` bigint NOT NULL COMMENT '用户ID', + `status` varchar(32) NOT NULL COMMENT '处理状态', + `failure_reason` varchar(512) NOT NULL DEFAULT '' COMMENT '失败原因', + `create_time` datetime(3) DEFAULT NULL COMMENT '创建时间', + `update_time` datetime(3) DEFAULT NULL COMMENT '更新时间', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_task_center_event_dedup` (`sys_origin`, `event_id`), + KEY `idx_task_center_event_dedup_status` (`status`), + KEY `idx_task_center_event_dedup_create_time` (`create_time`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='任务中心事件幂等表'; + +CREATE TABLE IF NOT EXISTS `task_center_event_archive_outbox` ( + `id` bigint NOT NULL COMMENT '主键ID', + `sys_origin` varchar(32) NOT NULL COMMENT '来源系统', + `event_id` varchar(128) NOT NULL COMMENT '业务事件ID', + `event_type` varchar(64) NOT NULL COMMENT '事件类型', + `user_id` bigint NOT NULL COMMENT '用户ID', + `occurred_at` datetime(3) DEFAULT NULL COMMENT '业务发生时间', + `raw_json` text NOT NULL COMMENT '待归档原始事件JSON', + `status` varchar(32) NOT NULL DEFAULT 'PENDING' COMMENT '归档状态', + `retry_count` int NOT NULL DEFAULT '0' COMMENT '重试次数', + `cos_key` varchar(512) NOT NULL DEFAULT '' COMMENT 'COS对象Key', + `failure_reason` varchar(512) NOT NULL DEFAULT '' COMMENT '失败原因', + `create_time` datetime(3) DEFAULT NULL COMMENT '创建时间', + `update_time` datetime(3) DEFAULT NULL COMMENT '更新时间', + `archived_at` datetime(3) DEFAULT NULL COMMENT '归档完成时间', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_task_center_archive_event` (`sys_origin`, `event_id`), + KEY `idx_task_center_archive_status_time` (`status`, `create_time`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='任务中心事件COS归档Outbox';