commit cfbc50a3fb2e98288197e137343c1862e67e5300 Author: zhx Date: Fri May 22 14:03:24 2026 +0800 init dashboard cdc worker diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..714be1f --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.DS_Store +/bin/ +/dist/ +/tmp/ +/.env +/.env.* +!/env.example diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..b32c75e --- /dev/null +++ b/Dockerfile @@ -0,0 +1,20 @@ +FROM --platform=$BUILDPLATFORM golang:1.24-alpine AS builder + +ARG TARGETOS +ARG TARGETARCH +ARG VERSION=dev + +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +RUN CGO_ENABLED=0 GOOS=${TARGETOS:-linux} GOARCH=${TARGETARCH:-amd64} \ + go build -trimpath -ldflags="-s -w -X main.version=${VERSION}" \ + -o /out/dashboard-cdc-worker ./cmd/dashboard-cdc-worker + +FROM alpine:3.22 +RUN apk add --no-cache ca-certificates tzdata wget +WORKDIR /application +COPY --from=builder /out/dashboard-cdc-worker /application/service +EXPOSE 2910 +ENTRYPOINT ["/application/service"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..e19e3f1 --- /dev/null +++ b/README.md @@ -0,0 +1,14 @@ +# dashboard-cdc-worker + +独立的国家大屏 CDC Worker，用 MySQL ROW binlog 读取业务事件，增量写入 `country_dashboard_*` 预聚合表，并可选写 Redis 今日实时 Hash。 + +## 关键约束 + +- 单独部署为 `dashboard-cdc-worker` 容器，不复用主 Go 服务。 +- 默认 `CDC_DRY_RUN=true`，上线验证链路时先只读 binlog 不写聚合表。 +- 只监听当前 Java 大屏口径依赖的表，避免扫源表。 +- MySQL 必须开启 `log_bin=ON`、`binlog_format=ROW`、`binlog_row_image=FULL`。 + +## 部署 + +镜像由 `ci/build-image.sh ` 构建。hy-app-monitor 会读取本仓库、构建镜像、更新 app-2 compose，并滚动重启 `dashboard-cdc-worker`。 diff --git a/ci/build-image.sh b/ci/build-image.sh new file mode 100755 index 0000000..63fad80 --- /dev/null +++ b/ci/build-image.sh @@ -0,0 +1,17 @@ +#!/usr/bin/env sh +set -eu + +if [ "$#" -lt 1 ]; then + echo "usage: $0 [version]" >&2 + exit 2 +fi + +IMAGE_REF="$1" +VERSION="${2:-dev}" +DOCKER_PLATFORM="${DOCKER_PLATFORM:-linux/amd64}" + +docker build \ + --platform "$DOCKER_PLATFORM" \ + --build-arg VERSION="$VERSION" \ + -t "$IMAGE_REF" \ + . diff --git a/cmd/dashboard-cdc-worker/main.go b/cmd/dashboard-cdc-worker/main.go new file mode 100644 index 0000000..344e2f6 --- /dev/null +++ b/cmd/dashboard-cdc-worker/main.go @@ -0,0 +1,99 @@ +package main + +import ( + "context" + "database/sql" + "errors" + "log/slog" + "os" + "os/signal" + "syscall" + "time" + + _ "github.com/go-sql-driver/mysql" + "github.com/redis/go-redis/v9" + + "dashboard-cdc-worker/internal/cdc" + "dashboard-cdc-worker/internal/config" + "dashboard-cdc-worker/internal/dashboard" + "dashboard-cdc-worker/internal/health" + "dashboard-cdc-worker/internal/storage" +) + +func main() { + logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})) + slog.SetDefault(logger) + + cfg, err := config.Load() + if err != nil { + logger.Error("load config failed", "error", err) + os.Exit(1) + } + + db, err := sql.Open("mysql", cfg.MySQL.AppDSN) + if err != nil { + logger.Error("open mysql failed", "error", err) + os.Exit(1) + } + defer db.Close() + db.SetMaxOpenConns(cfg.MySQL.MaxOpenConns) + db.SetMaxIdleConns(cfg.MySQL.MaxIdleConns) + db.SetConnMaxLifetime(5 * time.Minute) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + healthServer := health.New(cfg.Health.Addr) + healthServer.Start(logger) + defer func() { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _ = healthServer.Shutdown(shutdownCtx) + }() + + if err := db.PingContext(ctx); err != nil { + logger.Error("ping mysql failed", "error", err) + os.Exit(1) + } + + repo := storage.NewMySQLRepository(db, cfg) + if err := repo.EnsureSchema(ctx); err != nil { + logger.Error("ensure schema failed", "error", err) + os.Exit(1) + } + + var redisClient *redis.Client + if cfg.Redis.Enabled { + redisClient = redis.NewClient(&redis.Options{ + Addr: cfg.Redis.Addr, + Username: cfg.Redis.Username, + Password: cfg.Redis.Password, + DB: cfg.Redis.DB, + }) + defer redisClient.Close() + if err := redisClient.Ping(ctx).Err(); err != nil { + logger.Error("ping redis failed", "error", err) + os.Exit(1) + } + } + + projector := dashboard.NewProjector(cfg, repo, redisClient) + worker, err := cdc.NewWorker(cfg, repo, projector, logger) + if err != nil { + logger.Error("create cdc worker failed", "error", err) + os.Exit(1) + } + + logger.Info("dashboard cdc worker starting", + "dryRun", cfg.CDC.DryRun, + "startMode", cfg.CDC.StartMode, + "serverID", cfg.CDC.ServerID, + "statTimezones", cfg.Dashboard.StatTimezones, + "sysOrigins", cfg.Dashboard.SysOrigins, + ) + + if err := worker.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + logger.Error("dashboard cdc worker stopped with error", "error", err) + os.Exit(1) + } + logger.Info("dashboard cdc worker stopped") +} diff --git a/env.example b/env.example new file mode 100644 index 0000000..2757a07 --- /dev/null +++ b/env.example @@ -0,0 +1,13 @@ +CDC_MYSQL_ADDR=127.0.0.1:3306 +CDC_MYSQL_USER=dashboard_cdc +CDC_MYSQL_PASSWORD= +DASHBOARD_MYSQL_DSN=dashboard_cdc:password@tcp(127.0.0.1:3306)/likei?charset=utf8mb4&parseTime=true&loc=Asia%2FRiyadh +CDC_START_MODE=checkpoint +CDC_SERVER_ID=29301 +CDC_DRY_RUN=true +DASHBOARD_SYS_ORIGINS=LIKEI +DASHBOARD_STAT_TIMEZONES=Asia/Riyadh +DASHBOARD_STORAGE_TIMEZONE=Asia/Riyadh +REDIS_ENABLED=false +REDIS_ADDR=127.0.0.1:6379 +HEALTH_ADDR=:2910 diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..52ae970 --- /dev/null +++ b/go.mod @@ -0,0 +1,29 @@ +module dashboard-cdc-worker + +go 1.24.0 + +require ( + github.com/go-mysql-org/go-mysql v1.13.0 + 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 +) + +require ( + filippo.io/edwards25519 v1.1.0 // indirect + github.com/BurntSushi/toml v1.3.2 // indirect + 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/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 + 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/text v0.24.0 // indirect + gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..c0e63ff --- /dev/null +++ b/go.sum @@ -0,0 +1,89 @@ +filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= +filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= +github.com/BurntSushi/toml v1.3.2 h1:o7IhLm0Msx3BaB+n3Ag7L8EVlByGnpq14C4YWiu/gL8= +github.com/BurntSushi/toml v1.3.2/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ= +github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/go-mysql-org/go-mysql v1.13.0 h1:Hlsa5x1bX/wBFtMbdIOmb6YzyaVNBWnwrb8gSIEPMDc= +github.com/go-mysql-org/go-mysql v1.13.0/go.mod h1:FQxw17uRbFvMZFK+dPtIPufbU46nBdrGaxOw0ac9MFs= +github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo= +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/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= +github.com/klauspost/compress v1.17.8/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/pingcap/errors v0.11.0/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8= +github.com/pingcap/errors v0.11.5-0.20250318082626-8f80e5cb09ec h1:3EiGmeJWoNixU+EwllIn26x6s4njiWRXewdx2zlYa84= +github.com/pingcap/errors v0.11.5-0.20250318082626-8f80e5cb09ec/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg= +github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 h1:tdMsjOqUR7YXHoBitzdebTvOjs/swniBTOLy5XiMtuE= +github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86/go.mod h1:exzhVYca3WRtd6gclGNErRWb1qEgff3LYta0LvRmON4= +github.com/pingcap/log v1.1.1-0.20241212030209-7e3ff8601a2a h1:WIhmJBlNGmnCWH6TLMdZfNEDaiU8cFpZe3iaqDbQ0M8= +github.com/pingcap/log v1.1.1-0.20241212030209-7e3ff8601a2a/go.mod h1:ORfBOFp1eteu2odzsyaxI+b8TzJwgjwyQcGhI+9SfEA= +github.com/pingcap/tidb/pkg/parser v0.0.0-20250421232622-526b2c79173d h1:3Ej6eTuLZp25p3aH/EXdReRHY12hjZYs3RrGp7iLdag= +github.com/pingcap/tidb/pkg/parser v0.0.0-20250421232622-526b2c79173d/go.mod h1:+8feuexTKcXHZF/dkDfvCwEyBAmgb4paFc3/WeYV2eE= +github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.12.1 h1:k5iquqv27aBtnTm2tIkROUDp8JBXhXZIVu1InSgvovg= +github.com/redis/go-redis/v9 v9.12.1/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw= +github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= +github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= +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= +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= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= +go.uber.org/multierr v1.7.0/go.mod h1:7EAYxJLBy9rStEaz58O2t4Uvip6FSURkq8/ppBp95ak= +go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= +go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +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/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= +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/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +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-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/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= +gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= +gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/cdc/handler.go b/internal/cdc/handler.go new file mode 100644 index 0000000..284ad36 --- /dev/null +++ b/internal/cdc/handler.go @@ -0,0 +1,128 @@ +package cdc + +import ( + "context" + "fmt" + "log/slog" + "sync" + + "github.com/go-mysql-org/go-mysql/canal" + "github.com/go-mysql-org/go-mysql/mysql" + "github.com/go-mysql-org/go-mysql/replication" + + "dashboard-cdc-worker/internal/config" + "dashboard-cdc-worker/internal/dashboard" +) + +type handler struct { + canal.DummyEventHandler + cfg config.Config + store dashboard.Store + projector *dashboard.Projector + mapper *dashboard.Mapper + logger *slog.Logger + mu sync.RWMutex + file string +} + +func newHandler(cfg config.Config, store dashboard.Store, projector *dashboard.Projector, logger *slog.Logger) *handler { + return &handler{ + cfg: cfg, + store: store, + projector: projector, + mapper: dashboard.NewMapper(store, cfg.Dashboard.SysOrigins, cfg.Dashboard.UnknownCountryCode), + logger: logger, + } +} + +func (h *handler) OnRotate(_ *replication.EventHeader, rotateEvent *replication.RotateEvent) error { + h.setCurrentFile(string(rotateEvent.NextLogName)) + return nil +} + +func (h *handler) OnRow(e *canal.RowsEvent) error { + ctx := context.Background() + source := e.Table.Schema + "." + e.Table.Name + switch e.Action { + case canal.InsertAction: + for index, row := range e.Rows { + if err := h.applyRows(ctx, e, index, nil, row); err != nil { + return err + } + } + case canal.DeleteAction: + for index, row := range e.Rows { + if err := h.applyRows(ctx, e, index, row, nil); err != nil { + return err + } + } + case canal.UpdateAction: + for index := 0; index+1 < len(e.Rows); index += 2 { + if err := h.applyRows(ctx, e, index/2, e.Rows[index], e.Rows[index+1]); err != nil { + return err + } + } + default: + h.logger.Debug("ignore unsupported row action", "source", source, "action", e.Action) + } + return nil +} + +func (h *handler) OnPosSynced(_ *replication.EventHeader, pos mysql.Position, _ mysql.GTIDSet, force bool) error { + if h.cfg.CDC.DryRun { + return nil + } + if !force && h.cfg.CDC.FlushInterval > 0 { + // canal already throttles calls; keep this hook simple and always persist when invoked. + } + return h.store.SaveCheckpoint(context.Background(), pos) +} + +func (h *handler) String() string { + return "dashboard-cdc-worker" +} + +func (h *handler) setCurrentFile(file string) { + h.mu.Lock() + defer h.mu.Unlock() + h.file = file +} + +func (h *handler) currentFile() string { + h.mu.RLock() + defer h.mu.RUnlock() + return h.file +} + +func (h *handler) applyRows(ctx context.Context, e *canal.RowsEvent, rowIndex int, oldRaw []interface{}, newRaw []interface{}) error { + var contributions []dashboard.Contribution + var user *dashboard.UserCountry + source := e.Table.Schema + "." + e.Table.Name + + if oldRaw != nil { + oldRow := rowMap(e, oldRaw) + items, err := h.mapper.Map(ctx, e.Table.Schema, e.Table.Name, canal.DeleteAction, oldRow) + if err != nil { + return err + } + contributions = append(contributions, items...) + } + if newRaw != nil { + newRow := rowMap(e, newRaw) + if e.Table.Name == "user_base_info" { + u := dashboard.UserCountryFromRow(newRow, h.cfg.Dashboard.UnknownCountryCode) + user = &u + } + items, err := h.mapper.Map(ctx, e.Table.Schema, e.Table.Name, canal.InsertAction, newRow) + if err != nil { + return err + } + contributions = append(contributions, items...) + } + + eventKey := fmt.Sprintf("%s:%d:%s:%s:%d", h.currentFile(), e.Header.LogPos, source, e.Action, rowIndex) + if user != nil { + return h.projector.ApplyUser(ctx, eventKey, source, *user, contributions) + } + return h.projector.Apply(ctx, eventKey, source, contributions) +} diff --git a/internal/cdc/rows.go b/internal/cdc/rows.go new file mode 100644 index 0000000..6806f9d --- /dev/null +++ b/internal/cdc/rows.go @@ -0,0 +1,18 @@ +package cdc + +import ( + "github.com/go-mysql-org/go-mysql/canal" + + "dashboard-cdc-worker/internal/dashboard" +) + +func rowMap(e *canal.RowsEvent, row []interface{}) dashboard.Row { + mapped := make(dashboard.Row, len(e.Table.Columns)) + for index, column := range e.Table.Columns { + if index >= len(row) { + continue + } + mapped[column.Name] = row[index] + } + return mapped +} diff --git a/internal/cdc/worker.go b/internal/cdc/worker.go new file mode 100644 index 0000000..08e0d02 --- /dev/null +++ b/internal/cdc/worker.go @@ -0,0 +1,96 @@ +package cdc + +import ( + "context" + "fmt" + "log/slog" + "regexp" + "strings" + "time" + + "github.com/go-mysql-org/go-mysql/canal" + + "dashboard-cdc-worker/internal/config" + "dashboard-cdc-worker/internal/dashboard" +) + +type Worker struct { + cfg config.Config + store dashboard.Store + projector *dashboard.Projector + logger *slog.Logger + canal *canal.Canal +} + +func NewWorker(cfg config.Config, store dashboard.Store, projector *dashboard.Projector, logger *slog.Logger) (*Worker, error) { + canalCfg := canal.NewDefaultConfig() + canalCfg.Addr = cfg.MySQL.CDCAddr + canalCfg.User = cfg.MySQL.CDCUser + canalCfg.Password = cfg.MySQL.CDCPassword + canalCfg.ServerID = cfg.CDC.ServerID + canalCfg.Flavor = "mysql" + canalCfg.ParseTime = true + canalCfg.UseDecimal = true + if location, err := time.LoadLocation(cfg.Dashboard.StorageTimezone); err == nil { + canalCfg.TimestampStringLocation = location + } + canalCfg.HeartbeatPeriod = cfg.CDC.HeartbeatEvery + canalCfg.ReadTimeout = 90 * time.Second + canalCfg.IncludeTableRegex = includeRegex(cfg.CDC.IncludeTables) + canalCfg.Dump.ExecutionPath = "" + canalCfg.DiscardNoMetaRowEvent = true + canalCfg.Logger = logger + + instance, err := canal.NewCanal(canalCfg) + if err != nil { + return nil, err + } + return &Worker{ + cfg: cfg, + store: store, + projector: projector, + logger: logger, + canal: instance, + }, nil +} + +func (w *Worker) Run(ctx context.Context) error { + handler := newHandler(w.cfg, w.store, w.projector, w.logger) + w.canal.SetEventHandler(handler) + go func() { + <-ctx.Done() + w.canal.Close() + }() + + if w.cfg.CDC.StartMode == "checkpoint" { + pos, ok, err := w.store.LoadCheckpoint(ctx) + if err != nil { + return err + } + if ok && pos.Name != "" && pos.Pos > 0 { + handler.setCurrentFile(pos.Name) + w.logger.Info("starting cdc from checkpoint", "position", pos.String()) + return w.canal.RunFrom(pos) + } + } + + pos, err := w.canal.GetMasterPos() + if err != nil { + return fmt.Errorf("get master position: %w", err) + } + handler.setCurrentFile(pos.Name) + w.logger.Info("starting cdc from current master position", "position", pos.String()) + return w.canal.RunFrom(pos) +} + +func includeRegex(tables []string) []string { + out := make([]string, 0, len(tables)) + for _, table := range tables { + parts := strings.Split(strings.TrimSpace(table), ".") + if len(parts) != 2 || parts[0] == "" || parts[1] == "" { + continue + } + out = append(out, "^"+regexp.QuoteMeta(parts[0])+"\\."+regexp.QuoteMeta(parts[1])+"$") + } + return out +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..c371c84 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,239 @@ +package config + +import ( + "errors" + "fmt" + "net/url" + "os" + "strconv" + "strings" + "time" +) + +type Config struct { + MySQL MySQLConfig + Redis RedisConfig + CDC CDCConfig + Dashboard DashboardConfig + Health HealthConfig +} + +type MySQLConfig struct { + AppDSN string + CDCAddr string + CDCUser string + CDCPassword string + Schema string + WalletSchema string + MaxOpenConns int + MaxIdleConns int +} + +type RedisConfig struct { + Enabled bool + Addr string + Username string + Password string + DB int +} + +type CDCConfig struct { + ServerID uint32 + StartMode string + DryRun bool + IncludeTables []string + FlushInterval time.Duration + HeartbeatEvery time.Duration +} + +type DashboardConfig struct { + SysOrigins []string + StatTimezones []string + StorageTimezone string + RedisKeyPrefix string + RealtimeRedisTTL time.Duration + UnknownCountryCode string +} + +type HealthConfig struct { + Addr string +} + +func Load() (Config, error) { + cfg := Config{ + MySQL: MySQLConfig{ + AppDSN: env("DASHBOARD_MYSQL_DSN", env("MYSQL_DSN", "")), + CDCAddr: env("CDC_MYSQL_ADDR", ""), + CDCUser: env("CDC_MYSQL_USER", ""), + CDCPassword: env("CDC_MYSQL_PASSWORD", ""), + Schema: env("CDC_MYSQL_SCHEMA", "likei"), + WalletSchema: env("CDC_WALLET_SCHEMA", "likei_wallet"), + MaxOpenConns: envInt("MYSQL_MAX_OPEN_CONNS", 8), + MaxIdleConns: envInt("MYSQL_MAX_IDLE_CONNS", 4), + }, + Redis: RedisConfig{ + Enabled: envBool("REDIS_ENABLED", false), + Addr: env("REDIS_ADDR", ""), + Username: env("REDIS_USERNAME", ""), + Password: env("REDIS_PASSWORD", ""), + DB: envInt("REDIS_DB", 0), + }, + CDC: CDCConfig{ + ServerID: uint32(envInt("CDC_SERVER_ID", 29301)), + StartMode: strings.ToLower(env("CDC_START_MODE", "checkpoint")), + DryRun: envBool("CDC_DRY_RUN", true), + IncludeTables: csv(env("CDC_INCLUDE_TABLES", defaultTables())), + FlushInterval: envDuration("CDC_FLUSH_INTERVAL", 3*time.Second), + HeartbeatEvery: envDuration("CDC_HEARTBEAT_EVERY", 30*time.Second), + }, + Dashboard: DashboardConfig{ + SysOrigins: csv(env("DASHBOARD_SYS_ORIGINS", "LIKEI")), + StatTimezones: csv(env("DASHBOARD_STAT_TIMEZONES", "Asia/Riyadh")), + StorageTimezone: env("DASHBOARD_STORAGE_TIMEZONE", "Asia/Riyadh"), + RedisKeyPrefix: env("DASHBOARD_REDIS_KEY_PREFIX", "country_dashboard"), + RealtimeRedisTTL: envDuration("DASHBOARD_REALTIME_REDIS_TTL", 48*time.Hour), + UnknownCountryCode: env("DASHBOARD_UNKNOWN_COUNTRY_CODE", "UNKNOWN"), + }, + Health: HealthConfig{ + Addr: env("HEALTH_ADDR", ":2910"), + }, + } + + if cfg.MySQL.AppDSN == "" { + return Config{}, errors.New("DASHBOARD_MYSQL_DSN or MYSQL_DSN is required") + } + if cfg.MySQL.CDCAddr == "" || cfg.MySQL.CDCUser == "" { + return Config{}, errors.New("CDC_MYSQL_ADDR and CDC_MYSQL_USER are required") + } + if cfg.CDC.CDCPasswordRequired() && cfg.MySQL.CDCPassword == "" { + return Config{}, errors.New("CDC_MYSQL_PASSWORD is required") + } + if cfg.Redis.Enabled && cfg.Redis.Addr == "" { + return Config{}, errors.New("REDIS_ADDR is required when REDIS_ENABLED=true") + } + if cfg.CDC.ServerID == 0 { + return Config{}, errors.New("CDC_SERVER_ID must be greater than 0") + } + if cfg.CDC.StartMode != "checkpoint" && cfg.CDC.StartMode != "current" { + return Config{}, fmt.Errorf("unsupported CDC_START_MODE=%s", cfg.CDC.StartMode) + } + if len(cfg.Dashboard.SysOrigins) == 0 { + return Config{}, errors.New("DASHBOARD_SYS_ORIGINS is empty") + } + if len(cfg.Dashboard.StatTimezones) == 0 { + return Config{}, errors.New("DASHBOARD_STAT_TIMEZONES is empty") + } + return cfg, nil +} + +func (c CDCConfig) CDCPasswordRequired() bool { + return true +} + +func defaultTables() string { + return strings.Join([]string{ + "likei.user_base_info", + "likei.order_purchase_history", + "likei.order_user_purchase_pay", + "likei.game_lucky_gift_count", + "likei.game_lucky_box_count", + "likei.baishun_order_idempotency", + "likei.game_open_callback_log", + "likei.baishun_game_catalog", + "likei.lingxian_game_catalog", + "likei.sys_game_list_config", + "likei_wallet.wallet_gold_asset_record_1", + "likei_wallet.wallet_gold_asset_record_2", + "likei_wallet.wallet_gold_asset_record_3", + "likei_wallet.wallet_gold_asset_record_4", + "likei_wallet.wallet_gold_asset_record_5", + "likei_wallet.wallet_gold_asset_record_6", + "likei_wallet.wallet_gold_asset_record_7", + "likei_wallet.wallet_gold_asset_record_8", + "likei_wallet.wallet_gold_asset_record_9", + "likei_wallet.wallet_gold_asset_record_10", + "likei_wallet.wallet_gold_asset_record_11", + "likei_wallet.wallet_gold_asset_record_12", + "likei_wallet.user_freight_balance_running_water", + "likei_wallet.user_salary_account_running_water", + "likei_wallet.user_salary_diamond_running_water", + }, ",") +} + +func env(key string, fallback string) string { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + return value +} + +func envBool(key string, fallback bool) bool { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + parsed, err := strconv.ParseBool(value) + if err != nil { + return fallback + } + return parsed +} + +func envInt(key string, fallback int) int { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + parsed, err := strconv.Atoi(value) + if err != nil { + return fallback + } + return parsed +} + +func envDuration(key string, fallback time.Duration) time.Duration { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + parsed, err := time.ParseDuration(value) + if err == nil { + return parsed + } + seconds, err := strconv.Atoi(value) + if err != nil { + return fallback + } + return time.Duration(seconds) * time.Second +} + +func csv(value string) []string { + if strings.TrimSpace(value) == "" { + return nil + } + items := strings.Split(value, ",") + out := make([]string, 0, len(items)) + seen := make(map[string]struct{}, len(items)) + for _, item := range items { + item = strings.TrimSpace(item) + if item == "" { + continue + } + if _, exists := seen[item]; exists { + continue + } + seen[item] = struct{}{} + out = append(out, item) + } + return out +} + +func RedactedDSN(dsn string) string { + parsed, err := url.Parse("mysql://" + dsn) + if err != nil || parsed.User == nil { + return "mysql://***" + } + parsed.User = url.UserPassword(parsed.User.Username(), "***") + return strings.TrimPrefix(parsed.String(), "mysql://") +} diff --git a/internal/dashboard/mapper.go b/internal/dashboard/mapper.go new file mode 100644 index 0000000..74e9035 --- /dev/null +++ b/internal/dashboard/mapper.go @@ -0,0 +1,151 @@ +package dashboard + +import ( + "context" + "strings" + + "github.com/shopspring/decimal" +) + +type Mapper struct { + store Store + sysOrigins map[string]struct{} + unknownCountry string +} + +func NewMapper(store Store, sysOrigins []string, unknownCountry string) *Mapper { + origins := make(map[string]struct{}, len(sysOrigins)) + for _, origin := range sysOrigins { + origins[strings.TrimSpace(origin)] = struct{}{} + } + return &Mapper{store: store, sysOrigins: origins, unknownCountry: unknownCountry} +} + +func (m *Mapper) Map(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + switch table { + case "user_base_info": + return m.mapUser(ctx, schema, table, action, row) + case "order_purchase_history": + return m.mapOfficialRecharge(ctx, schema, table, action, row) + case "order_user_purchase_pay": + return m.mapMifapayRecharge(ctx, schema, table, action, row) + case "game_lucky_gift_count": + return m.mapLuckyGift(ctx, schema, table, action, row) + case "game_lucky_box_count": + return m.mapLuckyBox(ctx, schema, table, action, row) + case "user_freight_balance_running_water": + return m.mapDealerRecharge(ctx, schema, table, action, row) + case "user_salary_account_running_water", "user_salary_diamond_running_water": + return m.mapSalaryExchange(ctx, schema, table, action, row) + default: + if strings.HasPrefix(table, "wallet_gold_asset_record_") { + return m.mapWalletGame(ctx, schema, table, action, row) + } + return nil, nil + } +} + +func (m *Mapper) mapUser(_ context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + user := UserCountryFromRow(row, m.unknownCountry) + if !m.allowed(user.SysOrigin) || user.IsDeleted || user.UserID == 0 || user.CreatedAt.IsZero() { + return nil, nil + } + return []Contribution{{ + SourceSchema: schema, + SourceTable: table, + SourcePK: rowPK(row), + SourceAction: action, + DeltaSign: signForAction(action), + SysOrigin: user.SysOrigin, + UserID: user.UserID, + Country: user.Country, + EventTime: user.CreatedAt, + CountryNewUser: 1, + }}, nil +} + +func (m *Mapper) mapOfficialRecharge(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + if row.String("evn") != "PROD" || row.Bool("is_trial_period") || row.String("status") != "COMPLETE" { + return nil, nil + } + sysOrigin := row.String("sys_origin") + if !m.allowed(sysOrigin) { + return nil, nil + } + user, ok, err := m.user(ctx, row.Int64("user_id")) + if err != nil || !ok || user.IsDeleted { + return nil, err + } + if user.SysOrigin != sysOrigin { + return nil, nil + } + amount := row.Decimal("unit_price") + payPlatform := row.String("pay_platform") + c := m.base(schema, table, action, row, user, row.Time("create_time")) + if payPlatform != "MIFA_PAY" { + c.OfficialRecharge = amount + } + if payPlatform == "GOOGLE" { + c.GoogleRecharge = amount + } + c.NewUserRechargeCandidate = amount + return []Contribution{c}, nil +} + +func (m *Mapper) mapMifapayRecharge(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + if row.String("evn") != "PROD" || row.String("factory_code") != "MIFA_PAY" || + row.String("pay_status") != "SUCCESSFUL" || row.String("receipt_type") != "PAYMENT" || + row.String("refund_status") != "NONE" { + return nil, nil + } + return m.amountContribution(ctx, schema, table, action, row, row.Decimal("compute_usd_amount"), func(c *Contribution, amount decimal.Decimal) { + c.MifapayRecharge = amount + }) +} + +func (m *Mapper) mapDealerRecharge(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + if row.String("origin") != "PURCHASE" || row.Int64("type") != 0 || !row.Decimal("amount").GreaterThan(decimal.Zero) { + return nil, nil + } + return m.amountContribution(ctx, schema, table, action, row, row.Decimal("amount"), func(c *Contribution, amount decimal.Decimal) { + c.DealerRecharge = amount + }) +} + +func (m *Mapper) mapSalaryExchange(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + if row.Int64("type") != 1 { + return nil, nil + } + event := row.String("salary_event") + if event != "SALARY_EXCHANGE" && event != "SALARY_EXCHANGE_GOLD_COINS" && + event != "EXCHANGE_GOLD_COINS" && event != "BILL_EXCHANGE_GOLD_COINS" { + return nil, nil + } + return m.amountContribution(ctx, schema, table, action, row, row.Decimal("amount"), func(c *Contribution, amount decimal.Decimal) { + c.SalaryExchange = amount + }) +} + +func (m *Mapper) mapLuckyGift(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + return m.amountContribution(ctx, schema, table, action, row, row.Decimal("pay_amount"), func(c *Contribution, amount decimal.Decimal) { + c.LuckyGiftTotalFlow = amount + c.LuckyGiftPayout = row.Decimal("award_amount") + c.LuckyGiftUser = true + }) +} + +func (m *Mapper) mapLuckyBox(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + return m.amountContribution(ctx, schema, table, action, row, row.Decimal("pay_amount"), func(c *Contribution, amount decimal.Decimal) { + payout := row.Decimal("gift_amount") + c.GameTotalFlow = amount + c.GamePayout = payout + c.GameUser = true + c.GameProvider = "INTERNAL" + c.GameID = "LUCKY_BOX" + c.GameName = "Lucky Box" + c.GameConsume = amount + c.GamePayoutBy = payout + c.GameProfit = amount.Sub(payout) + c.GameOrder = 1 + }) +} diff --git a/internal/dashboard/mapper_helpers.go b/internal/dashboard/mapper_helpers.go new file mode 100644 index 0000000..617b0d1 --- /dev/null +++ b/internal/dashboard/mapper_helpers.go @@ -0,0 +1,98 @@ +package dashboard + +import ( + "context" + "strings" + "time" + + "github.com/shopspring/decimal" +) + +func (m *Mapper) amountContribution( + ctx context.Context, + schema string, + table string, + action string, + row Row, + amount decimal.Decimal, + apply func(*Contribution, decimal.Decimal), +) ([]Contribution, error) { + sysOrigin := row.String("sys_origin") + if !m.allowed(sysOrigin) { + return nil, nil + } + user, ok, err := m.user(ctx, row.Int64("user_id")) + if err != nil || !ok || user.IsDeleted { + return nil, err + } + if user.SysOrigin != sysOrigin { + return nil, nil + } + c := m.base(schema, table, action, row, user, row.Time("create_time")) + apply(&c, amount) + return []Contribution{c}, nil +} + +func (m *Mapper) base(schema string, table string, action string, row Row, user UserCountry, eventTime time.Time) Contribution { + return Contribution{ + SourceSchema: schema, + SourceTable: table, + SourcePK: rowPK(row), + SourceAction: action, + DeltaSign: signForAction(action), + SysOrigin: user.SysOrigin, + UserID: user.UserID, + Country: user.Country, + EventTime: eventTime, + UserCreatedAt: user.CreatedAt, + } +} + +func (m *Mapper) user(ctx context.Context, userID int64) (UserCountry, bool, error) { + if userID == 0 { + return UserCountry{}, false, nil + } + return m.store.UserCountry(ctx, userID) +} + +func (m *Mapper) allowed(sysOrigin string) bool { + if len(m.sysOrigins) == 0 { + return true + } + _, ok := m.sysOrigins[sysOrigin] + return ok +} + +func UserCountryFromRow(row Row, unknownCountry string) UserCountry { + code := strings.TrimSpace(row.String("country_code")) + if code == "" { + code = unknownCountry + } + name := strings.TrimSpace(row.String("country_name")) + if name == "" { + name = code + } + return UserCountry{ + UserID: row.Int64("id"), + SysOrigin: strings.TrimSpace(row.String("origin_sys")), + Country: Country{Code: code, Name: name}, + IsDeleted: row.Bool("is_del"), + CreatedAt: row.Time("create_time"), + UpdatedAt: row.Time("update_time"), + } +} + +func rowPK(row Row) string { + id := strings.TrimSpace(row.String("id")) + if id != "" { + return id + } + return strings.TrimSpace(row.String("event_id")) +} + +func signForAction(action string) int { + if action == "delete" { + return -1 + } + return 1 +} diff --git a/internal/dashboard/mapper_wallet.go b/internal/dashboard/mapper_wallet.go new file mode 100644 index 0000000..3c39cd1 --- /dev/null +++ b/internal/dashboard/mapper_wallet.go @@ -0,0 +1,79 @@ +package dashboard + +import ( + "context" + "strings" + + "github.com/shopspring/decimal" +) + +var walletGameEvents = map[string]struct{}{ + "GAME_SLOT_MACHINE": {}, "GAME_BARBECUE": {}, "TURNTABLE_GAME": {}, "LUDO_GAME": {}, + "GAME_BURST_CRYSTAL": {}, "EGG": {}, "FRUIT_GAME": {}, "DOUBLE_LAYER_FRUIT": {}, + "DOUBLE_LAYER_BARBECUE": {}, "GAME_FRUIT_BET": {}, "GAME_FRUIT_AWARD": {}, "GAME_BACK": {}, + "GAME_WINNING": {}, "LUDO_VICTORY": {}, "LUDO_REFUND": {}, "GAME_LUDO_REFUND": {}, + "USD_GAME_WIN": {}, "USD_GAME_REFUND": {}, "USD_GAME_REFUND_DRAW": {}, "TEEN_PATTI": {}, + "BET_TEEN_PATTI": {}, "LOTTERY_NUMBER_BET": {}, "LOTTERY_REWARD": {}, +} + +func (m *Mapper) mapWalletGame(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) { + eventType := row.String("event_type") + if !isWalletGameEvent(eventType) { + return nil, nil + } + sysOrigin := row.String("sys_origin") + if !m.allowed(sysOrigin) { + return nil, nil + } + user, ok, err := m.user(ctx, row.Int64("user_id")) + if err != nil || !ok || user.IsDeleted { + return nil, err + } + if user.SysOrigin != sysOrigin { + return nil, nil + } + + provider, gameID, gameName, err := m.store.ResolveWalletGame(ctx, sysOrigin, eventType, row.String("event_id")) + if err != nil { + return nil, err + } + amount := row.Decimal("penny_amount").Div(decimal.NewFromInt(100)) + consume := decimal.Zero + payout := decimal.Zero + if row.Int64("type") == 1 { + consume = amount + } else if row.Int64("type") == 0 { + payout = amount + } + + c := m.base(schema, table, action, row, user, row.MillisTime("create_time")) + c.GameTotalFlow = consume + c.GamePayout = payout + c.GameUser = true + c.GameProvider = provider + c.GameID = gameID + c.GameName = gameName + c.GameConsume = consume + c.GamePayoutBy = payout + c.GameProfit = consume.Sub(payout) + c.GameOrder = 1 + return []Contribution{c}, nil +} + +func isWalletGameEvent(eventType string) bool { + if _, ok := walletGameEvents[eventType]; ok { + return true + } + if eventType == "LUCKY_GIFT" || eventType == "LUCKY_GIFT_GOLD_REWARD" || + eventType == "LUCKY_BOX_GAME" || eventType == "NEW_LUCKY_BOX_GAME" || eventType == "GAME_RAKE" { + return false + } + if strings.HasPrefix(eventType, "GIVE_GIFT") || strings.HasPrefix(eventType, "ACCEPT_GIFT") { + return false + } + return strings.HasPrefix(eventType, "BAISHUN_GAME") || + strings.HasPrefix(eventType, "GAME_OPEN_") || + strings.HasPrefix(eventType, "YOMI_GAME") || + strings.HasPrefix(eventType, "HKYS_GAME") || + strings.HasPrefix(eventType, "HOT_GAME") +} diff --git a/internal/dashboard/period.go b/internal/dashboard/period.go new file mode 100644 index 0000000..168c49d --- /dev/null +++ b/internal/dashboard/period.go @@ -0,0 +1,95 @@ +package dashboard + +import ( + "fmt" + "time" +) + +func PeriodWindows(eventTime time.Time, statZone *time.Location) []PeriodWindow { + localDate := startOfDay(eventTime.In(statZone)) + dayEnd := localDate.AddDate(0, 0, 1) + + weekday := int(localDate.Weekday()) + if weekday == 0 { + weekday = 7 + } + weekStart := localDate.AddDate(0, 0, -(weekday - 1)) + weekEnd := weekStart.AddDate(0, 0, 7) + + monthStart := time.Date(localDate.Year(), localDate.Month(), 1, 0, 0, 0, 0, statZone) + monthEnd := monthStart.AddDate(0, 1, 0) + + _, isoWeek := localDate.ISOWeek() + isoYear, _ := localDate.ISOWeek() + + return []PeriodWindow{ + { + Type: PeriodDay, + Key: localDate.Format("2006-01-02"), + Name: localDate.Format("2006-01-02"), + StartDate: ptrTime(localDate), + EndDate: ptrTime(dayEnd), + }, + { + Type: PeriodWeek, + Key: fmt.Sprintf("%04d-W%02d", isoYear, isoWeek), + Name: fmt.Sprintf("%04d-W%02d", isoYear, isoWeek), + StartDate: ptrTime(weekStart), + EndDate: ptrTime(weekEnd), + }, + { + Type: PeriodMonth, + Key: localDate.Format("2006-01"), + Name: localDate.Format("2006-01"), + StartDate: ptrTime(monthStart), + EndDate: ptrTime(monthEnd), + }, + { + Type: PeriodAll, + Key: PeriodAll, + Name: "全部", + }, + } +} + +func SamePeriod(left time.Time, right time.Time, window PeriodWindow, statZone *time.Location) bool { + if left.IsZero() || right.IsZero() { + return false + } + if window.Type == PeriodAll { + return true + } + leftLocal := left.In(statZone) + rightLocal := right.In(statZone) + switch window.Type { + case PeriodDay: + return startOfDay(leftLocal).Equal(startOfDay(rightLocal)) + case PeriodWeek: + leftYear, leftWeek := leftLocal.ISOWeek() + rightYear, rightWeek := rightLocal.ISOWeek() + return leftYear == rightYear && leftWeek == rightWeek + case PeriodMonth: + return leftLocal.Year() == rightLocal.Year() && leftLocal.Month() == rightLocal.Month() + default: + return false + } +} + +func DailyDate(eventTime time.Time, storageZone *time.Location) time.Time { + return startOfDay(eventTime.In(storageZone)) +} + +func DateNumber(day time.Time) int { + year, month, date := day.Date() + return year*10000 + int(month)*100 + date +} + +func ptrTime(value time.Time) *time.Time { + v := value + return &v +} + +func startOfDay(value time.Time) time.Time { + year, month, date := value.Date() + return time.Date(year, month, date, 0, 0, 0, 0, value.Location()) +} diff --git a/internal/dashboard/projector.go b/internal/dashboard/projector.go new file mode 100644 index 0000000..e5de125 --- /dev/null +++ b/internal/dashboard/projector.go @@ -0,0 +1,102 @@ +package dashboard + +import ( + "context" + "database/sql" + "log/slog" + "time" + + "github.com/redis/go-redis/v9" + + "dashboard-cdc-worker/internal/config" +) + +type Projector struct { + cfg config.Config + store Store + redisClient *redis.Client + storageZone *time.Location + statZones map[string]*time.Location +} + +func NewProjector(cfg config.Config, store Store, redisClient *redis.Client) *Projector { + storageZone := mustLocation(cfg.Dashboard.StorageTimezone) + statZones := make(map[string]*time.Location, len(cfg.Dashboard.StatTimezones)) + for _, zone := range cfg.Dashboard.StatTimezones { + statZones[zone] = mustLocation(zone) + } + return &Projector{ + cfg: cfg, + store: store, + redisClient: redisClient, + storageZone: storageZone, + statZones: statZones, + } +} + +func (p *Projector) Apply(ctx context.Context, eventKey string, source string, contributions []Contribution) error { + return p.apply(ctx, eventKey, source, nil, contributions) +} + +func (p *Projector) ApplyUser(ctx context.Context, eventKey string, source string, user UserCountry, contributions []Contribution) error { + return p.apply(ctx, eventKey, source, &user, contributions) +} + +func (p *Projector) apply(ctx context.Context, eventKey string, source string, user *UserCountry, contributions []Contribution) error { + if user == nil && len(contributions) == 0 { + return nil + } + if len(contributions) == 0 { + contributions = nil + } + if p.cfg.CDC.DryRun { + slog.Info("dashboard cdc dry run", "eventKey", eventKey, "source", source, "user", user != nil, "contributions", len(contributions)) + return nil + } + if err := p.store.WithTx(ctx, func(tx *sql.Tx) error { + applied, err := p.store.MarkEventApplied(ctx, tx, eventKey, source) + if err != nil { + return err + } + if !applied { + return nil + } + if user != nil { + if err := p.store.UpsertUserCountry(ctx, tx, *user); err != nil { + return err + } + } + for _, contribution := range contributions { + if contribution.Empty() { + continue + } + if contribution.DeltaSign == 0 { + contribution.DeltaSign = 1 + } + if err := p.store.ApplyDailyContribution(ctx, tx, contribution, p.cfg.Dashboard.StorageTimezone); err != nil { + return err + } + for statTimezone := range p.statZones { + if err := p.store.ApplyPeriodContribution(ctx, tx, contribution, statTimezone); err != nil { + return err + } + if err := p.store.ApplyGamePeriodContribution(ctx, tx, contribution, statTimezone); err != nil { + return err + } + } + } + return nil + }); err != nil { + return err + } + return p.writeRealtimeRedis(ctx, contributions) +} + +func mustLocation(name string) *time.Location { + location, err := time.LoadLocation(name) + if err != nil { + slog.Warn("invalid timezone, fallback to UTC", "timezone", name, "error", err) + return time.UTC + } + return location +} diff --git a/internal/dashboard/redis.go b/internal/dashboard/redis.go new file mode 100644 index 0000000..ff390a2 --- /dev/null +++ b/internal/dashboard/redis.go @@ -0,0 +1,48 @@ +package dashboard + +import ( + "context" + "fmt" + + "github.com/redis/go-redis/v9" + "github.com/shopspring/decimal" +) + +func (p *Projector) writeRealtimeRedis(ctx context.Context, contributions []Contribution) error { + if p.redisClient == nil || len(contributions) == 0 { + return nil + } + pipe := p.redisClient.Pipeline() + for _, c := range contributions { + if c.Empty() || c.DeltaSign == 0 { + continue + } + key := fmt.Sprintf("%s:today:%s:%s:%s", + p.cfg.Dashboard.RedisKeyPrefix, + DailyDate(c.EventTime, p.storageZone).Format("2006-01-02"), + c.SysOrigin, + c.Country.Code, + ) + sign := decimal.NewFromInt(int64(c.DeltaSign)) + hincrDecimal(pipe, ctx, key, "official_recharge", c.OfficialRecharge.Mul(sign)) + hincrDecimal(pipe, ctx, key, "mifapay_recharge", c.MifapayRecharge.Mul(sign)) + hincrDecimal(pipe, ctx, key, "google_recharge", c.GoogleRecharge.Mul(sign)) + hincrDecimal(pipe, ctx, key, "dealer_recharge", c.DealerRecharge.Mul(sign)) + hincrDecimal(pipe, ctx, key, "salary_exchange", c.SalaryExchange.Mul(sign)) + hincrDecimal(pipe, ctx, key, "lucky_gift_total_flow", c.LuckyGiftTotalFlow.Mul(sign)) + hincrDecimal(pipe, ctx, key, "lucky_gift_payout", c.LuckyGiftPayout.Mul(sign)) + hincrDecimal(pipe, ctx, key, "game_total_flow", c.GameTotalFlow.Mul(sign)) + hincrDecimal(pipe, ctx, key, "game_payout", c.GamePayout.Mul(sign)) + pipe.Expire(ctx, key, p.cfg.Dashboard.RealtimeRedisTTL) + } + _, err := pipe.Exec(ctx) + return err +} + +func hincrDecimal(pipe redis.Pipeliner, ctx context.Context, key string, field string, value decimal.Decimal) { + if value.IsZero() { + return + } + floatValue, _ := value.Float64() + pipe.HIncrByFloat(ctx, key, field, floatValue) +} diff --git a/internal/dashboard/row.go b/internal/dashboard/row.go new file mode 100644 index 0000000..292f383 --- /dev/null +++ b/internal/dashboard/row.go @@ -0,0 +1,141 @@ +package dashboard + +import ( + "fmt" + "strconv" + "strings" + "time" + + "github.com/shopspring/decimal" +) + +type Row map[string]interface{} + +func (r Row) String(name string) string { + value, ok := r[name] + if !ok || value == nil { + return "" + } + switch typed := value.(type) { + case []byte: + return string(typed) + case string: + return typed + default: + return fmt.Sprint(typed) + } +} + +func (r Row) Int64(name string) int64 { + value, ok := r[name] + if !ok || value == nil { + return 0 + } + switch typed := value.(type) { + case int: + return int64(typed) + case int8: + return int64(typed) + case int16: + return int64(typed) + case int32: + return int64(typed) + case int64: + return typed + case uint: + return int64(typed) + case uint8: + return int64(typed) + case uint16: + return int64(typed) + case uint32: + return int64(typed) + case uint64: + return int64(typed) + case []byte: + parsed, _ := strconv.ParseInt(string(typed), 10, 64) + return parsed + case string: + parsed, _ := strconv.ParseInt(typed, 10, 64) + return parsed + default: + parsed, _ := strconv.ParseInt(fmt.Sprint(typed), 10, 64) + return parsed + } +} + +func (r Row) Bool(name string) bool { + return r.Int64(name) != 0 +} + +func (r Row) Decimal(name string) decimal.Decimal { + value, ok := r[name] + if !ok || value == nil { + return decimal.Zero + } + switch typed := value.(type) { + case decimal.Decimal: + return typed + case float32: + return decimal.NewFromFloat32(typed) + case float64: + return decimal.NewFromFloat(typed) + case int64: + return decimal.NewFromInt(typed) + case int: + return decimal.NewFromInt(int64(typed)) + case []byte: + parsed, _ := decimal.NewFromString(string(typed)) + return parsed + case string: + parsed, _ := decimal.NewFromString(typed) + return parsed + default: + parsed, _ := decimal.NewFromString(fmt.Sprint(typed)) + return parsed + } +} + +func (r Row) Time(name string) time.Time { + value, ok := r[name] + if !ok || value == nil { + return time.Time{} + } + switch typed := value.(type) { + case time.Time: + return typed + case []byte: + return parseTimeString(string(typed)) + case string: + return parseTimeString(typed) + default: + return parseTimeString(fmt.Sprint(typed)) + } +} + +func (r Row) MillisTime(name string) time.Time { + millis := r.Int64(name) + if millis <= 0 { + return time.Time{} + } + return time.UnixMilli(millis) +} + +func parseTimeString(value string) time.Time { + text := strings.TrimSpace(value) + if text == "" || strings.EqualFold(text, "") { + return time.Time{} + } + for _, layout := range []string{ + time.RFC3339Nano, + "2006-01-02 15:04:05.999999", + "2006-01-02 15:04:05", + "2006-01-02", + } { + parsed, err := time.ParseInLocation(layout, text, time.Local) + if err == nil { + return parsed + } + } + return time.Time{} +} diff --git a/internal/dashboard/store.go b/internal/dashboard/store.go new file mode 100644 index 0000000..48a791c --- /dev/null +++ b/internal/dashboard/store.go @@ -0,0 +1,23 @@ +package dashboard + +import ( + "context" + "database/sql" + + "github.com/go-mysql-org/go-mysql/mysql" +) + +type Store interface { + WithTx(ctx context.Context, fn func(*sql.Tx) error) error + MarkEventApplied(ctx context.Context, tx *sql.Tx, eventKey string, source string) (bool, error) + SaveCheckpoint(ctx context.Context, pos mysql.Position) error + LoadCheckpoint(ctx context.Context) (mysql.Position, bool, error) + + UpsertUserCountry(ctx context.Context, tx *sql.Tx, user UserCountry) error + UserCountry(ctx context.Context, userID int64) (UserCountry, bool, error) + ResolveWalletGame(ctx context.Context, sysOrigin string, eventType string, eventID string) (string, string, string, error) + + ApplyDailyContribution(ctx context.Context, tx *sql.Tx, c Contribution, storageZone string) error + ApplyPeriodContribution(ctx context.Context, tx *sql.Tx, c Contribution, statTimezone string) error + ApplyGamePeriodContribution(ctx context.Context, tx *sql.Tx, c Contribution, statTimezone string) error +} diff --git a/internal/dashboard/types.go b/internal/dashboard/types.go new file mode 100644 index 0000000..8e7cb15 --- /dev/null +++ b/internal/dashboard/types.go @@ -0,0 +1,99 @@ +package dashboard + +import ( + "time" + + "github.com/shopspring/decimal" +) + +const ( + PeriodDay = "DAY" + PeriodWeek = "WEEK" + PeriodMonth = "MONTH" + PeriodAll = "ALL" + + MetricCountryNewUser = "COUNTRY_NEW_USER" + MetricLuckyGiftUser = "LUCKY_GIFT_USER" + MetricGameUser = "GAME_USER" +) + +type Country struct { + Code string + Name string +} + +type Contribution struct { + SourceSchema string + SourceTable string + SourcePK string + SourceAction string + EventKey string + DeltaSign int + + SysOrigin string + UserID int64 + Country Country + EventTime time.Time + UserCreatedAt time.Time + + CountryNewUser int64 + NewUserRecharge decimal.Decimal + NewUserRechargeCandidate decimal.Decimal + OfficialRecharge decimal.Decimal + MifapayRecharge decimal.Decimal + GoogleRecharge decimal.Decimal + DealerRecharge decimal.Decimal + SalaryExchange decimal.Decimal + LuckyGiftTotalFlow decimal.Decimal + LuckyGiftUser bool + LuckyGiftPayout decimal.Decimal + GameTotalFlow decimal.Decimal + GameUser bool + GamePayout decimal.Decimal + + GameProvider string + GameID string + GameName string + GameConsume decimal.Decimal + GamePayoutBy decimal.Decimal + GameProfit decimal.Decimal + GameOrder int64 +} + +func (c Contribution) Empty() bool { + return c.CountryNewUser == 0 && + c.NewUserRecharge.IsZero() && + c.NewUserRechargeCandidate.IsZero() && + c.OfficialRecharge.IsZero() && + c.MifapayRecharge.IsZero() && + c.GoogleRecharge.IsZero() && + c.DealerRecharge.IsZero() && + c.SalaryExchange.IsZero() && + c.LuckyGiftTotalFlow.IsZero() && + !c.LuckyGiftUser && + c.LuckyGiftPayout.IsZero() && + c.GameTotalFlow.IsZero() && + !c.GameUser && + c.GamePayout.IsZero() && + c.GameConsume.IsZero() && + c.GamePayoutBy.IsZero() && + c.GameProfit.IsZero() && + c.GameOrder == 0 +} + +type PeriodWindow struct { + Type string + Key string + Name string + StartDate *time.Time + EndDate *time.Time +} + +type UserCountry struct { + UserID int64 + SysOrigin string + Country Country + IsDeleted bool + CreatedAt time.Time + UpdatedAt time.Time +} diff --git a/internal/health/server.go b/internal/health/server.go new file mode 100644 index 0000000..c6da19a --- /dev/null +++ b/internal/health/server.go @@ -0,0 +1,35 @@ +package health + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + "time" +) + +type Server struct { + server *http.Server +} + +func New(addr string) *Server { + mux := http.NewServeMux() + mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]string{"code": "ok"}) + }) + return &Server{server: &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second}} +} + +func (s *Server) Start(logger *slog.Logger) { + go func() { + if err := s.server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + logger.Error("health server failed", "error", err) + } + }() +} + +func (s *Server) Shutdown(ctx context.Context) error { + return s.server.Shutdown(ctx) +} diff --git a/internal/storage/aggregate.go b/internal/storage/aggregate.go new file mode 100644 index 0000000..3cd1e4e --- /dev/null +++ b/internal/storage/aggregate.go @@ -0,0 +1,181 @@ +package storage + +import ( + "context" + "database/sql" + "time" + + "github.com/shopspring/decimal" + + "dashboard-cdc-worker/internal/dashboard" +) + +func (r *MySQLRepository) ApplyDailyContribution(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, storageZone string) error { + zone, err := time.LoadLocation(storageZone) + if err != nil { + zone = time.UTC + } + day := dashboard.DailyDate(c.EventTime, zone) + newUserRecharge := decimal.Zero + if dashboard.SamePeriod(c.UserCreatedAt, c.EventTime, dashboard.PeriodWindow{Type: dashboard.PeriodDay}, zone) { + newUserRecharge = c.NewUserRechargeCandidate + } + if err := r.upsertDailyAmounts(ctx, tx, c, day, newUserRecharge); err != nil { + return err + } + if c.CountryNewUser > 0 { + if err := r.applyDailyUserMetric(ctx, tx, c, day, dashboard.MetricCountryNewUser); err != nil { + return err + } + } + if c.LuckyGiftUser { + if err := r.applyDailyUserMetric(ctx, tx, c, day, dashboard.MetricLuckyGiftUser); err != nil { + return err + } + } + if c.GameUser { + if err := r.applyDailyUserMetric(ctx, tx, c, day, dashboard.MetricGameUser); err != nil { + return err + } + } + return nil +} + +func (r *MySQLRepository) ApplyPeriodContribution(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, statTimezone string) error { + zone, err := time.LoadLocation(statTimezone) + if err != nil { + zone = time.UTC + } + for _, window := range dashboard.PeriodWindows(c.EventTime, zone) { + newUserRecharge := decimal.Zero + if dashboard.SamePeriod(c.UserCreatedAt, c.EventTime, window, zone) { + newUserRecharge = c.NewUserRechargeCandidate + } + if err := r.upsertPeriodAmounts(ctx, tx, c, window, statTimezone, newUserRecharge); err != nil { + return err + } + if c.CountryNewUser > 0 { + if err := r.applyPeriodUserMetric(ctx, tx, c, window, statTimezone, dashboard.MetricCountryNewUser); err != nil { + return err + } + } + if c.LuckyGiftUser { + if err := r.applyPeriodUserMetric(ctx, tx, c, window, statTimezone, dashboard.MetricLuckyGiftUser); err != nil { + return err + } + } + if c.GameUser { + if err := r.applyPeriodUserMetric(ctx, tx, c, window, statTimezone, dashboard.MetricGameUser); err != nil { + return err + } + } + } + return nil +} + +func (r *MySQLRepository) ApplyGamePeriodContribution(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, statTimezone string) error { + if c.GameProvider == "" || c.GameID == "" { + return nil + } + zone, err := time.LoadLocation(statTimezone) + if err != nil { + zone = time.UTC + } + for _, window := range dashboard.PeriodWindows(c.EventTime, zone) { + if err := r.upsertGamePeriodAmounts(ctx, tx, c, window, statTimezone); err != nil { + return err + } + if c.GameUser { + if err := r.applyGamePeriodUserMetric(ctx, tx, c, window, statTimezone); err != nil { + return err + } + } + } + return nil +} + +func (r *MySQLRepository) upsertDailyAmounts(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, day time.Time, newUserRecharge decimal.Decimal) error { + sign := decimal.NewFromInt(int64(c.DeltaSign)) + _, err := tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_daily_metric ( + stat_date, date_number, sys_origin, country_code, country_name, + new_user_recharge, official_recharge, mifapay_recharge, google_recharge, + dealer_recharge, salary_exchange, lucky_gift_total_flow, lucky_gift_payout, + game_total_flow, game_payout, refreshed_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE + country_name = VALUES(country_name), + new_user_recharge = new_user_recharge + VALUES(new_user_recharge), + official_recharge = official_recharge + VALUES(official_recharge), + mifapay_recharge = mifapay_recharge + VALUES(mifapay_recharge), + google_recharge = google_recharge + VALUES(google_recharge), + dealer_recharge = dealer_recharge + VALUES(dealer_recharge), + salary_exchange = salary_exchange + VALUES(salary_exchange), + lucky_gift_total_flow = lucky_gift_total_flow + VALUES(lucky_gift_total_flow), + lucky_gift_payout = lucky_gift_payout + VALUES(lucky_gift_payout), + game_total_flow = game_total_flow + VALUES(game_total_flow), + game_payout = game_payout + VALUES(game_payout), + refreshed_at = NOW() +`, day, dashboard.DateNumber(day), c.SysOrigin, c.Country.Code, c.Country.Name, + newUserRecharge.Mul(sign), c.OfficialRecharge.Mul(sign), c.MifapayRecharge.Mul(sign), c.GoogleRecharge.Mul(sign), + c.DealerRecharge.Mul(sign), c.SalaryExchange.Mul(sign), c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign), + c.GameTotalFlow.Mul(sign), c.GamePayout.Mul(sign)) + return err +} + +func (r *MySQLRepository) upsertPeriodAmounts(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string, newUserRecharge decimal.Decimal) error { + sign := decimal.NewFromInt(int64(c.DeltaSign)) + _, err := tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_period_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, new_user_recharge, official_recharge, mifapay_recharge, + google_recharge, dealer_recharge, salary_exchange, lucky_gift_total_flow, lucky_gift_payout, + game_total_flow, game_payout, refreshed_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE + period_name = VALUES(period_name), + period_start_date = VALUES(period_start_date), + period_end_date = VALUES(period_end_date), + country_name = VALUES(country_name), + new_user_recharge = new_user_recharge + VALUES(new_user_recharge), + official_recharge = official_recharge + VALUES(official_recharge), + mifapay_recharge = mifapay_recharge + VALUES(mifapay_recharge), + google_recharge = google_recharge + VALUES(google_recharge), + dealer_recharge = dealer_recharge + VALUES(dealer_recharge), + salary_exchange = salary_exchange + VALUES(salary_exchange), + lucky_gift_total_flow = lucky_gift_total_flow + VALUES(lucky_gift_total_flow), + lucky_gift_payout = lucky_gift_payout + VALUES(lucky_gift_payout), + game_total_flow = game_total_flow + VALUES(game_total_flow), + game_payout = game_payout + VALUES(game_payout), + refreshed_at = NOW() +`, window.Type, window.Key, statTimezone, window.Name, window.StartDate, window.EndDate, + c.SysOrigin, c.Country.Code, c.Country.Name, newUserRecharge.Mul(sign), c.OfficialRecharge.Mul(sign), + c.MifapayRecharge.Mul(sign), c.GoogleRecharge.Mul(sign), c.DealerRecharge.Mul(sign), c.SalaryExchange.Mul(sign), + c.LuckyGiftTotalFlow.Mul(sign), c.LuckyGiftPayout.Mul(sign), c.GameTotalFlow.Mul(sign), c.GamePayout.Mul(sign)) + return err +} + +func (r *MySQLRepository) upsertGamePeriodAmounts(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string) error { + sign := decimal.NewFromInt(int64(c.DeltaSign)) + _, err := tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_game_period_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, game_provider, game_id, game_name, + consume_amount, payout_amount, profit_amount, order_count, refreshed_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE + period_name = VALUES(period_name), + period_start_date = VALUES(period_start_date), + period_end_date = VALUES(period_end_date), + country_name = VALUES(country_name), + game_name = VALUES(game_name), + consume_amount = consume_amount + VALUES(consume_amount), + payout_amount = payout_amount + VALUES(payout_amount), + profit_amount = profit_amount + VALUES(profit_amount), + order_count = order_count + VALUES(order_count), + refreshed_at = NOW() +`, window.Type, window.Key, statTimezone, window.Name, window.StartDate, window.EndDate, + c.SysOrigin, c.Country.Code, c.Country.Name, c.GameProvider, c.GameID, c.GameName, + c.GameConsume.Mul(sign), c.GamePayoutBy.Mul(sign), c.GameProfit.Mul(sign), c.GameOrder*int64(c.DeltaSign)) + return err +} diff --git a/internal/storage/game.go b/internal/storage/game.go new file mode 100644 index 0000000..6785b76 --- /dev/null +++ b/internal/storage/game.go @@ -0,0 +1,161 @@ +package storage + +import ( + "context" + "database/sql" + "strconv" + "strings" +) + +func (r *MySQLRepository) ResolveWalletGame(ctx context.Context, sysOrigin string, eventType string, eventID string) (string, string, string, error) { + if provider, gameID, gameName, ok, err := r.resolveBaishunByWalletEvent(ctx, sysOrigin, eventID); ok || err != nil { + return provider, gameID, gameName, err + } + if provider, gameID, gameName, ok, err := r.resolveGameOpenByWalletEvent(ctx, sysOrigin, eventType, eventID); ok || err != nil { + return provider, gameID, gameName, err + } + + provider, gameID, fallback := fallbackGameMeta(eventType) + gameName, err := r.lookupGameName(ctx, sysOrigin, provider, gameID) + if err != nil { + return provider, gameID, fallback, err + } + if gameName == "" { + gameName = fallback + } + return provider, gameID, gameName, nil +} + +func (r *MySQLRepository) resolveBaishunByWalletEvent(ctx context.Context, sysOrigin string, eventID string) (string, string, string, bool, error) { + if strings.TrimSpace(eventID) == "" { + return "", "", "", false, nil + } + var vendorID int64 + err := r.db.QueryRowContext(ctx, ` +SELECT vendor_game_id +FROM baishun_order_idempotency +WHERE sys_origin = ? AND wallet_event_id = ? +LIMIT 1 +`, sysOrigin, eventID).Scan(&vendorID) + if err == sql.ErrNoRows { + return "", "", "", false, nil + } + if err != nil { + return "", "", "", false, err + } + gameID := strconv.FormatInt(vendorID, 10) + gameName, err := r.lookupGameName(ctx, sysOrigin, "BAISHUN", gameID) + if gameName == "" { + gameName = gameID + } + return "BAISHUN", gameID, gameName, true, err +} + +func (r *MySQLRepository) resolveGameOpenByWalletEvent(ctx context.Context, sysOrigin string, eventType string, eventID string) (string, string, string, bool, error) { + orderID := strings.TrimPrefix(eventID, "GAME_OPEN:") + if orderID == eventID && !strings.HasPrefix(eventType, "GAME_OPEN_") { + return "", "", "", false, nil + } + if orderID != "" && orderID != eventID { + var gameID string + err := r.db.QueryRowContext(ctx, ` +SELECT game_id +FROM game_open_callback_log +WHERE sys_origin = ? AND order_id = ? +LIMIT 1 +`, sysOrigin, orderID).Scan(&gameID) + if err != nil && err != sql.ErrNoRows { + return "", "", "", false, err + } + if err == nil && strings.TrimSpace(gameID) != "" { + gameName, err := r.lookupGameName(ctx, sysOrigin, "GAME_OPEN", gameID) + if gameName == "" { + gameName = gameID + } + return "GAME_OPEN", gameID, gameName, true, err + } + } + if strings.HasPrefix(eventType, "GAME_OPEN_") { + gameID := strings.TrimPrefix(eventType, "GAME_OPEN_") + gameName, err := r.lookupGameName(ctx, sysOrigin, "GAME_OPEN", gameID) + if gameName == "" { + gameName = gameID + } + return "GAME_OPEN", gameID, gameName, true, err + } + return "", "", "", false, nil +} + +func (r *MySQLRepository) lookupGameName(ctx context.Context, sysOrigin string, provider string, gameID string) (string, error) { + var name string + if provider == "BAISHUN" { + err := r.db.QueryRowContext(ctx, ` +SELECT name +FROM baishun_game_catalog +WHERE sys_origin = ? AND profile = 'PROD' AND vendor_game_id = ? +LIMIT 1 +`, sysOrigin, gameID).Scan(&name) + if err == nil || err != sql.ErrNoRows { + return name, err + } + } + err := r.db.QueryRowContext(ctx, ` +SELECT name +FROM lingxian_game_catalog +WHERE sys_origin = ? AND profile = 'PROD' AND vendor_game_id = ? +LIMIT 1 +`, sysOrigin, gameID).Scan(&name) + if err == nil || err != sql.ErrNoRows { + return name, err + } + err = r.db.QueryRowContext(ctx, ` +SELECT name +FROM sys_game_list_config +WHERE sys_origin = ? AND game_id = ? +LIMIT 1 +`, sysOrigin, gameID).Scan(&name) + if err == sql.ErrNoRows { + return "", nil + } + return name, err +} + +func fallbackGameMeta(eventType string) (string, string, string) { + switch { + case strings.HasPrefix(eventType, "BAISHUN_GAME_"): + return "BAISHUN", strings.TrimPrefix(eventType, "BAISHUN_GAME_"), strings.TrimPrefix(eventType, "BAISHUN_GAME_") + case strings.HasPrefix(eventType, "YOMI_GAME_"): + return "YOMI", strings.TrimPrefix(eventType, "YOMI_GAME_"), strings.TrimPrefix(eventType, "YOMI_GAME_") + case strings.HasPrefix(eventType, "HKYS_GAME_"): + return "HKYS", strings.TrimPrefix(eventType, "HKYS_GAME_"), strings.TrimPrefix(eventType, "HKYS_GAME_") + case strings.HasPrefix(eventType, "HOT_GAME_"): + return "HOT", strings.TrimPrefix(eventType, "HOT_GAME_"), strings.TrimPrefix(eventType, "HOT_GAME_") + case in(eventType, "FRUIT_GAME", "DOUBLE_LAYER_FRUIT", "GAME_FRUIT_BET", "GAME_FRUIT_AWARD"): + return "INTERNAL", "FRUIT_GAME", "Fruit Game" + case in(eventType, "LUDO_GAME", "LUDO_VICTORY", "LUDO_REFUND", "GAME_LUDO_REFUND"): + return "INTERNAL", "LUDO_GAME", "Ludo Game" + case in(eventType, "GAME_BARBECUE", "DOUBLE_LAYER_BARBECUE"): + return "INTERNAL", "BARBECUE_GAME", "Barbecue Game" + case eventType == "GAME_SLOT_MACHINE": + return "INTERNAL", "SLOT_MACHINE", "Slot Machine" + case eventType == "TURNTABLE_GAME": + return "INTERNAL", "TURNTABLE_GAME", "Turntable Game" + case eventType == "GAME_BURST_CRYSTAL": + return "INTERNAL", "BURST_CRYSTAL", "Burst Crystal" + case in(eventType, "TEEN_PATTI", "BET_TEEN_PATTI"): + return "INTERNAL", "TEEN_PATTI", "Teen Patti" + case in(eventType, "LOTTERY_NUMBER_BET", "LOTTERY_REWARD"): + return "INTERNAL", "LOTTERY_NUMBER", "Lottery Number" + default: + return "INTERNAL", eventType, eventType + } +} + +func in(value string, candidates ...string) bool { + for _, candidate := range candidates { + if value == candidate { + return true + } + } + return false +} diff --git a/internal/storage/repository.go b/internal/storage/repository.go new file mode 100644 index 0000000..a38c30b --- /dev/null +++ b/internal/storage/repository.go @@ -0,0 +1,87 @@ +package storage + +import ( + "context" + "database/sql" + "fmt" + + "github.com/go-mysql-org/go-mysql/mysql" + + "dashboard-cdc-worker/internal/config" +) + +const serviceName = "dashboard-cdc-worker" + +type MySQLRepository struct { + db *sql.DB + cfg config.Config +} + +func NewMySQLRepository(db *sql.DB, cfg config.Config) *MySQLRepository { + return &MySQLRepository{db: db, cfg: cfg} +} + +func (r *MySQLRepository) WithTx(ctx context.Context, fn func(*sql.Tx) error) error { + tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + return err + } + if err := fn(tx); err != nil { + _ = tx.Rollback() + return err + } + return tx.Commit() +} + +func (r *MySQLRepository) LoadCheckpoint(ctx context.Context) (mysql.Position, bool, error) { + var file string + var pos uint32 + err := r.db.QueryRowContext(ctx, ` +SELECT binlog_file, binlog_pos +FROM dashboard_cdc_checkpoint +WHERE service_name = ? +`, serviceName).Scan(&file, &pos) + if err == sql.ErrNoRows { + return mysql.Position{}, false, nil + } + if err != nil { + return mysql.Position{}, false, err + } + return mysql.Position{Name: file, Pos: pos}, true, nil +} + +func (r *MySQLRepository) SaveCheckpoint(ctx context.Context, pos mysql.Position) error { + _, err := r.db.ExecContext(ctx, ` +INSERT INTO dashboard_cdc_checkpoint (service_name, binlog_file, binlog_pos, updated_at) +VALUES (?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE + binlog_file = VALUES(binlog_file), + binlog_pos = VALUES(binlog_pos), + updated_at = NOW() +`, serviceName, pos.Name, pos.Pos) + return err +} + +func (r *MySQLRepository) MarkEventApplied(ctx context.Context, tx *sql.Tx, eventKey string, source string) (bool, error) { + result, err := tx.ExecContext(ctx, ` +INSERT IGNORE INTO dashboard_cdc_event_dedup (event_key, source, created_at) +VALUES (?, ?, NOW()) +`, eventKey, source) + if err != nil { + return false, err + } + affected, err := result.RowsAffected() + if err != nil { + return false, err + } + return affected > 0, nil +} + +func execAll(ctx context.Context, db *sql.DB, statements []string) error { + for _, stmt := range statements { + if _, err := db.ExecContext(ctx, stmt); err != nil { + return fmt.Errorf("exec schema failed: %w; sql=%s", err, stmt) + } + } + return nil +} diff --git a/internal/storage/schema.go b/internal/storage/schema.go new file mode 100644 index 0000000..a7d4d69 --- /dev/null +++ b/internal/storage/schema.go @@ -0,0 +1,159 @@ +package storage + +import "context" + +func (r *MySQLRepository) EnsureSchema(ctx context.Context) error { + return execAll(ctx, r.db, []string{ + `CREATE TABLE IF NOT EXISTS dashboard_cdc_checkpoint ( + service_name varchar(64) NOT NULL, + binlog_file varchar(255) NOT NULL, + binlog_pos int unsigned NOT NULL, + updated_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (service_name) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS dashboard_cdc_event_dedup ( + event_key varchar(255) NOT NULL, + source varchar(128) NOT NULL, + created_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (event_key), + KEY idx_dashboard_cdc_event_dedup_created (created_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS dashboard_cdc_user_country ( + user_id bigint NOT NULL, + sys_origin varchar(50) NOT NULL, + country_code varchar(30) NOT NULL, + country_name varchar(100) NOT NULL, + is_deleted tinyint(1) NOT NULL DEFAULT 0, + user_created_at datetime NULL, + 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) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + createDailyMetricTable, + createPeriodMetricTable, + createPeriodUserMetricTable, + createGamePeriodMetricTable, + createGamePeriodUserMetricTable, + }) +} + +const createDailyMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_daily_metric ( + id bigint NOT NULL AUTO_INCREMENT, + stat_date date NOT NULL, + date_number int NOT NULL, + sys_origin varchar(50) NOT NULL, + country_code varchar(64) NOT NULL, + country_name varchar(128) NOT NULL, + country_new_user bigint NOT NULL DEFAULT 0, + new_user_recharge decimal(20,2) NOT NULL DEFAULT 0, + official_recharge decimal(20,2) NOT NULL DEFAULT 0, + mifapay_recharge decimal(20,2) NOT NULL DEFAULT 0, + google_recharge decimal(20,2) NOT NULL DEFAULT 0, + dealer_recharge decimal(20,2) NOT NULL DEFAULT 0, + salary_exchange decimal(20,2) NOT NULL DEFAULT 0, + lucky_gift_total_flow decimal(20,2) NOT NULL DEFAULT 0, + lucky_gift_user bigint NOT NULL DEFAULT 0, + lucky_gift_payout decimal(20,2) NOT NULL DEFAULT 0, + game_total_flow decimal(20,2) NOT NULL DEFAULT 0, + game_user bigint NOT NULL DEFAULT 0, + game_payout decimal(20,2) NOT NULL DEFAULT 0, + refreshed_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uk_daily_origin_country (stat_date, sys_origin, country_code), + KEY idx_daily_origin_date (sys_origin, stat_date) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci` + +const createPeriodMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_period_metric ( + id bigint NOT NULL AUTO_INCREMENT, + period_type varchar(16) NOT NULL, + period_key varchar(32) NOT NULL, + stat_timezone varchar(64) NOT NULL DEFAULT 'Asia/Riyadh', + period_name varchar(64) NOT NULL, + period_start_date date NULL, + period_end_date date NULL, + sys_origin varchar(50) NOT NULL, + country_code varchar(64) NOT NULL, + country_name varchar(128) NOT NULL, + country_new_user bigint NOT NULL DEFAULT 0, + new_user_recharge decimal(20,2) NOT NULL DEFAULT 0, + official_recharge decimal(20,2) NOT NULL DEFAULT 0, + mifapay_recharge decimal(20,2) NOT NULL DEFAULT 0, + google_recharge decimal(20,2) NOT NULL DEFAULT 0, + dealer_recharge decimal(20,2) NOT NULL DEFAULT 0, + salary_exchange decimal(20,2) NOT NULL DEFAULT 0, + lucky_gift_total_flow decimal(20,2) NOT NULL DEFAULT 0, + lucky_gift_user bigint NOT NULL DEFAULT 0, + lucky_gift_payout decimal(20,2) NOT NULL DEFAULT 0, + game_total_flow decimal(20,2) NOT NULL DEFAULT 0, + game_user bigint NOT NULL DEFAULT 0, + game_payout decimal(20,2) NOT NULL DEFAULT 0, + refreshed_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uk_period_origin_country (period_type, period_key, stat_timezone, sys_origin, country_code), + KEY idx_origin_period (sys_origin, stat_timezone, period_type, period_key) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci` + +const createPeriodUserMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_period_user_metric ( + id bigint NOT NULL AUTO_INCREMENT, + period_type varchar(16) NOT NULL, + period_key varchar(32) NOT NULL, + stat_timezone varchar(64) NOT NULL DEFAULT 'Asia/Riyadh', + period_name varchar(64) NOT NULL, + period_start_date date NULL, + period_end_date date NULL, + sys_origin varchar(50) NOT NULL, + country_code varchar(64) NOT NULL, + country_name varchar(128) NOT NULL, + user_id bigint NOT NULL, + metric_type varchar(32) NOT NULL, + created_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uk_period_user_metric (period_type, period_key, stat_timezone, sys_origin, country_code, user_id, metric_type), + KEY idx_origin_user_metric (sys_origin, stat_timezone, user_id, metric_type) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci` + +const createGamePeriodMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_game_period_metric ( + id bigint NOT NULL AUTO_INCREMENT, + period_type varchar(16) NOT NULL, + period_key varchar(32) NOT NULL, + stat_timezone varchar(64) NOT NULL DEFAULT 'Asia/Riyadh', + period_name varchar(64) NOT NULL, + period_start_date date NULL, + period_end_date date NULL, + sys_origin varchar(50) NOT NULL, + country_code varchar(64) NOT NULL, + country_name varchar(128) NOT NULL, + game_provider varchar(32) NOT NULL, + game_id varchar(128) NOT NULL, + game_name varchar(256) NOT NULL, + consume_amount decimal(20,2) NOT NULL DEFAULT 0, + payout_amount decimal(20,2) NOT NULL DEFAULT 0, + profit_amount decimal(20,2) NOT NULL DEFAULT 0, + user_count bigint NOT NULL DEFAULT 0, + order_count bigint NOT NULL DEFAULT 0, + refreshed_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uk_game_period (sys_origin, stat_timezone, period_type, period_key, country_code, game_provider, game_id), + KEY idx_game_period_game (sys_origin, stat_timezone, period_type, period_key, game_provider, game_id, country_code) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci` + +const createGamePeriodUserMetricTable = `CREATE TABLE IF NOT EXISTS country_dashboard_game_period_user_metric ( + id bigint NOT NULL AUTO_INCREMENT, + period_type varchar(16) NOT NULL, + period_key varchar(32) NOT NULL, + stat_timezone varchar(64) NOT NULL DEFAULT 'Asia/Riyadh', + period_name varchar(64) NOT NULL, + period_start_date date NULL, + period_end_date date NULL, + sys_origin varchar(50) NOT NULL, + country_code varchar(64) NOT NULL, + country_name varchar(128) NOT NULL, + game_provider varchar(32) NOT NULL, + game_id varchar(128) NOT NULL, + user_id bigint NOT NULL, + created_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uk_game_period_user (period_type, period_key, stat_timezone, sys_origin, country_code, game_provider, game_id, user_id), + KEY idx_game_user (sys_origin, stat_timezone, user_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci` diff --git a/internal/storage/user.go b/internal/storage/user.go new file mode 100644 index 0000000..a95332d --- /dev/null +++ b/internal/storage/user.go @@ -0,0 +1,95 @@ +package storage + +import ( + "context" + "database/sql" + "time" + + "dashboard-cdc-worker/internal/dashboard" +) + +func (r *MySQLRepository) UpsertUserCountry(ctx context.Context, tx *sql.Tx, user dashboard.UserCountry) error { + _, err := tx.ExecContext(ctx, ` +INSERT INTO dashboard_cdc_user_country ( + user_id, sys_origin, country_code, country_name, is_deleted, user_created_at, user_updated_at, updated_at +) VALUES (?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE + sys_origin = VALUES(sys_origin), + country_code = VALUES(country_code), + country_name = VALUES(country_name), + is_deleted = VALUES(is_deleted), + user_created_at = VALUES(user_created_at), + user_updated_at = VALUES(user_updated_at), + updated_at = NOW() +`, user.UserID, user.SysOrigin, user.Country.Code, user.Country.Name, user.IsDeleted, nullTime(user.CreatedAt), nullTime(user.UpdatedAt)) + return err +} + +func (r *MySQLRepository) UserCountry(ctx context.Context, userID int64) (dashboard.UserCountry, bool, error) { + var user dashboard.UserCountry + var countryCode, countryName string + var isDeleted bool + var createdAt, updatedAt sql.NullTime + err := r.db.QueryRowContext(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 = ? +`, userID).Scan(&user.UserID, &user.SysOrigin, &countryCode, &countryName, &isDeleted, &createdAt, &updatedAt) + if err == sql.ErrNoRows { + return r.userCountryFromSource(ctx, userID) + } + if err != nil { + return dashboard.UserCountry{}, false, err + } + user.Country = dashboard.Country{Code: countryCode, Name: countryName} + user.IsDeleted = isDeleted + if createdAt.Valid { + user.CreatedAt = createdAt.Time + } + if updatedAt.Valid { + user.UpdatedAt = updatedAt.Time + } + return user, true, nil +} + +func (r *MySQLRepository) userCountryFromSource(ctx context.Context, userID int64) (dashboard.UserCountry, bool, error) { + var user dashboard.UserCountry + var countryCode, countryName string + var isDeleted bool + var createdAt, updatedAt sql.NullTime + err := r.db.QueryRowContext(ctx, ` +SELECT id, origin_sys, IFNULL(NULLIF(country_code, ''), ?), IFNULL(NULLIF(country_name, ''), IFNULL(NULLIF(country_code, ''), ?)), + IFNULL(is_del, 0), create_time, update_time +FROM user_base_info +WHERE id = ? +`, r.cfg.Dashboard.UnknownCountryCode, r.cfg.Dashboard.UnknownCountryCode, userID). + Scan(&user.UserID, &user.SysOrigin, &countryCode, &countryName, &isDeleted, &createdAt, &updatedAt) + if err == sql.ErrNoRows { + return dashboard.UserCountry{}, false, nil + } + if err != nil { + return dashboard.UserCountry{}, false, err + } + user.Country = dashboard.Country{Code: countryCode, Name: countryName} + user.IsDeleted = isDeleted + if createdAt.Valid { + user.CreatedAt = createdAt.Time + } + if updatedAt.Valid { + user.UpdatedAt = updatedAt.Time + } + _, _ = r.db.ExecContext(ctx, ` +INSERT INTO dashboard_cdc_user_country ( + user_id, sys_origin, country_code, country_name, is_deleted, user_created_at, user_updated_at, updated_at +) VALUES (?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE updated_at = updated_at +`, user.UserID, user.SysOrigin, user.Country.Code, user.Country.Name, user.IsDeleted, nullTime(user.CreatedAt), nullTime(user.UpdatedAt)) + return user, true, nil +} + +func nullTime(value time.Time) interface{} { + if value.IsZero() { + return nil + } + return value +} diff --git a/internal/storage/user_metrics.go b/internal/storage/user_metrics.go new file mode 100644 index 0000000..f5ab07b --- /dev/null +++ b/internal/storage/user_metrics.go @@ -0,0 +1,177 @@ +package storage + +import ( + "context" + "database/sql" + "time" + + "dashboard-cdc-worker/internal/dashboard" +) + +func (r *MySQLRepository) applyDailyUserMetric(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, day time.Time, metricType string) error { + if c.DeltaSign < 0 { + return r.deleteDailyUserMetric(ctx, tx, c, day, metricType) + } + inserted, err := insertPeriodUserMetric(ctx, tx, c, dashboard.PeriodWindow{ + Type: dashboard.PeriodDay, + Key: day.Format("2006-01-02"), + Name: day.Format("2006-01-02"), + StartDate: dashboardPtr(day), + EndDate: dashboardPtr(day.AddDate(0, 0, 1)), + }, r.cfg.Dashboard.StorageTimezone, metricType) + if err != nil || !inserted { + return err + } + column := dailyMetricColumn(metricType) + if column == "" { + return nil + } + _, err = tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_daily_metric ( + stat_date, date_number, sys_origin, country_code, country_name, refreshed_at +) VALUES (?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE `+column+` = `+column+` + 1, country_name = VALUES(country_name), refreshed_at = NOW() +`, day, dashboard.DateNumber(day), c.SysOrigin, c.Country.Code, c.Country.Name) + return err +} + +func (r *MySQLRepository) applyPeriodUserMetric(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string, metricType string) error { + if c.DeltaSign < 0 { + return r.deletePeriodUserMetric(ctx, tx, c, window, statTimezone, metricType) + } + inserted, err := insertPeriodUserMetric(ctx, tx, c, window, statTimezone, metricType) + if err != nil || !inserted { + return err + } + column := periodMetricColumn(metricType) + if column == "" { + return nil + } + _, err = tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_period_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, refreshed_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE `+column+` = `+column+` + 1, country_name = VALUES(country_name), refreshed_at = NOW() +`, window.Type, window.Key, statTimezone, window.Name, window.StartDate, window.EndDate, + c.SysOrigin, c.Country.Code, c.Country.Name) + return err +} + +func (r *MySQLRepository) deleteDailyUserMetric(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, day time.Time, metricType string) error { + window := dashboard.PeriodWindow{ + Type: dashboard.PeriodDay, + Key: day.Format("2006-01-02"), + Name: day.Format("2006-01-02"), + StartDate: dashboardPtr(day), + EndDate: dashboardPtr(day.AddDate(0, 0, 1)), + } + deleted, err := deletePeriodUserMetric(ctx, tx, c, window, r.cfg.Dashboard.StorageTimezone, metricType) + if err != nil || !deleted { + return err + } + column := dailyMetricColumn(metricType) + if column == "" { + return nil + } + _, err = tx.ExecContext(ctx, ` +UPDATE country_dashboard_daily_metric +SET `+column+` = GREATEST(`+column+` - 1, 0), refreshed_at = NOW() +WHERE stat_date = ? AND sys_origin = ? AND country_code = ? +`, day, c.SysOrigin, c.Country.Code) + return err +} + +func (r *MySQLRepository) deletePeriodUserMetric(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string, metricType string) error { + deleted, err := deletePeriodUserMetric(ctx, tx, c, window, statTimezone, metricType) + if err != nil || !deleted { + return err + } + column := periodMetricColumn(metricType) + if column == "" { + return nil + } + _, err = tx.ExecContext(ctx, ` +UPDATE country_dashboard_period_metric +SET `+column+` = GREATEST(`+column+` - 1, 0), refreshed_at = NOW() +WHERE period_type = ? AND period_key = ? AND stat_timezone = ? AND sys_origin = ? AND country_code = ? +`, window.Type, window.Key, statTimezone, c.SysOrigin, c.Country.Code) + return err +} + +func (r *MySQLRepository) applyGamePeriodUserMetric(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string) error { + result, err := tx.ExecContext(ctx, ` +INSERT IGNORE INTO country_dashboard_game_period_user_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, game_provider, game_id, user_id +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +`, window.Type, window.Key, statTimezone, window.Name, window.StartDate, window.EndDate, + c.SysOrigin, c.Country.Code, c.Country.Name, c.GameProvider, c.GameID, c.UserID) + if err != nil { + return err + } + affected, err := result.RowsAffected() + if err != nil || affected == 0 { + return err + } + _, err = tx.ExecContext(ctx, ` +INSERT INTO country_dashboard_game_period_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, game_provider, game_id, game_name, refreshed_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()) +ON DUPLICATE KEY UPDATE user_count = user_count + 1, country_name = VALUES(country_name), + game_name = VALUES(game_name), refreshed_at = NOW() +`, window.Type, window.Key, statTimezone, window.Name, window.StartDate, window.EndDate, + c.SysOrigin, c.Country.Code, c.Country.Name, c.GameProvider, c.GameID, c.GameName) + return err +} + +func insertPeriodUserMetric(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string, metricType string) (bool, error) { + result, err := tx.ExecContext(ctx, ` +INSERT IGNORE INTO country_dashboard_period_user_metric ( + period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date, + sys_origin, country_code, country_name, user_id, metric_type +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +`, window.Type, window.Key, statTimezone, window.Name, window.StartDate, window.EndDate, + c.SysOrigin, c.Country.Code, c.Country.Name, c.UserID, metricType) + if err != nil { + return false, err + } + affected, err := result.RowsAffected() + return affected > 0, err +} + +func deletePeriodUserMetric(ctx context.Context, tx *sql.Tx, c dashboard.Contribution, window dashboard.PeriodWindow, statTimezone string, metricType string) (bool, error) { + result, err := tx.ExecContext(ctx, ` +DELETE FROM country_dashboard_period_user_metric +WHERE period_type = ? AND period_key = ? AND stat_timezone = ? + AND sys_origin = ? AND country_code = ? AND user_id = ? AND metric_type = ? +`, window.Type, window.Key, statTimezone, c.SysOrigin, c.Country.Code, c.UserID, metricType) + if err != nil { + return false, err + } + affected, err := result.RowsAffected() + return affected > 0, err +} + +func dailyMetricColumn(metricType string) string { + switch metricType { + case dashboard.MetricCountryNewUser: + return "country_new_user" + case dashboard.MetricLuckyGiftUser: + return "lucky_gift_user" + case dashboard.MetricGameUser: + return "game_user" + default: + return "" + } +} + +func periodMetricColumn(metricType string) string { + return dailyMetricColumn(metricType) +} + +func dashboardPtr(value time.Time) *time.Time { + v := value + return &v +} diff --git a/migrations/001_dashboard_cdc_worker.sql b/migrations/001_dashboard_cdc_worker.sql new file mode 100644 index 0000000..a1eb4b6 --- /dev/null +++ b/migrations/001_dashboard_cdc_worker.sql @@ -0,0 +1,28 @@ +CREATE TABLE IF NOT EXISTS dashboard_cdc_checkpoint ( + service_name varchar(64) NOT NULL, + binlog_file varchar(255) NOT NULL, + binlog_pos int unsigned NOT NULL, + updated_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (service_name) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS dashboard_cdc_event_dedup ( + event_key varchar(255) NOT NULL, + source varchar(128) NOT NULL, + created_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (event_key), + KEY idx_dashboard_cdc_event_dedup_created (created_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS dashboard_cdc_user_country ( + user_id bigint NOT NULL, + sys_origin varchar(50) NOT NULL, + country_code varchar(30) NOT NULL, + country_name varchar(100) NOT NULL, + is_deleted tinyint(1) NOT NULL DEFAULT 0, + user_created_at datetime NULL, + 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) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;