3015 lines
94 KiB
Markdown
3015 lines
94 KiB
Markdown
# statisticService 看板 + 事件采集 实施计划
|
||
|
||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||
|
||
**Goal:** 实现 `statisticService` 后端服务（Go Dubbo-go），提供数据看板 7 RPC + 事件采集框架（TrackEvent / BatchTrackEvent），并将 5 个业务服务（social / asset / gallery / task / user）改造接入 TrackEvent 上报。
|
||
|
||
**Architecture:**
|
||
- `statisticService`：Go Dubbo-go 服务，端口 20009，暴露 7 个看板 RPC（经 gateway）+ 2 个事件 RPC（内部调用，TrackEvent / BatchTrackEvent）
|
||
- 事件流：业务服务 → Dubbo TrackEvent → ChannelEventSink → event_flusher Worker 批量落库到 `statistic.events` 表（按日分区）
|
||
- 看板数据：4 个物化视图（MV）+ 3 个预聚表（Worker 维护），看板查询走 MV/预聚表，Redis 5min TTL 缓存
|
||
- Gateway：经 `backend/gateway/controller/statistic_controller.go` 暴露 7 个 GET `/api/v1/dashboard/*` 路由，响应统一包装 `{code:200, data:resp}`
|
||
|
||
**Tech Stack:** Go (dubbo-go v3 triple)、GORM、PostgreSQL (按日分区 + 物化视图 + 预聚表)、Redis 缓存、Prometheus 指标、gin (gateway)
|
||
|
||
**Spec:** `docs/superpowers/specs/2026-06-04-statistic-service-design.md`（含 §0.1 本期实施范围）
|
||
|
||
---
|
||
|
||
## 文件结构
|
||
|
||
### 新建
|
||
|
||
```
|
||
backend/
|
||
├── proto/
|
||
│ ├── event.proto # 新建：通用事件类型（独立 proto）
|
||
│ └── statistic.proto # 新建：statistic 服务 + 7 看板 RPC + 2 事件 RPC
|
||
├── pkg/proto/
|
||
│ ├── event/ # 新建：event.proto 编译产物
|
||
│ └── statistic/ # 新建：statistic.proto 编译产物
|
||
├── services/statisticService/ # 新建（目录已存在但为空）
|
||
│ ├── go.mod / go.sum # 新建
|
||
│ ├── main.go # 新建
|
||
│ ├── config/
|
||
│ │ └── statistic_config.go # 新建
|
||
│ ├── model/
|
||
│ │ ├── event.go # 新建：Event 模型
|
||
│ │ └── metric.go # 新建：7 看板响应结构
|
||
│ ├── repository/
|
||
│ │ ├── event_repo.go # 新建：events 表 + 3 预聚表
|
||
│ │ └── dashboard_repo.go # 新建：4 MV 聚合查询
|
||
│ ├── sink/
|
||
│ │ ├── event_sink.go # 新建：EventSink 接口
|
||
│ │ └── channel_sink.go # 新建：ChannelEventSink（本期）
|
||
│ ├── service/
|
||
│ │ ├── event_service.go # 新建：事件校验 + Submit
|
||
│ │ ├── dashboard_service.go # 新建：7 看板 RPC 业务逻辑
|
||
│ │ ├── metric_service.go # 新建：MV 刷新协调
|
||
│ │ └── cache.go # 新建：Redis 缓存封装
|
||
│ ├── provider/
|
||
│ │ ├── statistic_mobile_provider.go # 新建：7 看板 RPC（mobile）
|
||
│ │ └── statistic_internal_provider.go # 新建：TrackEvent / BatchTrackEvent（internal）
|
||
│ ├── worker/
|
||
│ │ ├── event_flusher.go # 新建：事件批量落库
|
||
│ │ ├── metric_recent_level_ups_updater.go # 新建：同步触发
|
||
│ │ ├── metric_upcoming_level_ups_updater.go # 新建：15min ticker
|
||
│ │ ├── metric_weekly_user_income_updater.go # 新建：5min ticker
|
||
│ │ ├── materializer.go # 新建：4 MV 刷新
|
||
│ │ └── partitioner.go # 新建：events 分区管理
|
||
│ ├── handler/
|
||
│ │ └── healthz.go # 新建：健康检查
|
||
│ ├── metrics/
|
||
│ │ └── metrics.go # 新建：Prometheus 指标声明
|
||
│ ├── client/ # 新建：跨服务 RPC 客户端
|
||
│ │ ├── user_rpc_client.go
|
||
│ │ └── gallery_rpc_client.go
|
||
│ └── scripts/ # 新建：测试 fixture
|
||
│ └── testhelper/
|
||
├── migrations/
|
||
│ ├── 2026_06_08_001_statistic_events.sql
|
||
│ ├── 2026_06_08_002_statistic_mv_daily_user_income.sql
|
||
│ ├── 2026_06_08_003_statistic_mv_daily_exhibition_revenue.sql
|
||
│ ├── 2026_06_08_004_statistic_mv_daily_like_income.sql
|
||
│ ├── 2026_06_08_005_statistic_mv_asset_level_distribution.sql
|
||
│ ├── 2026_06_08_006_statistic_metric_weekly_user_income.sql
|
||
│ ├── 2026_06_08_007_statistic_metric_recent_level_ups.sql
|
||
│ ├── 2026_06_08_008_statistic_metric_upcoming_level_ups.sql
|
||
│ ├── 2026_06_08_009_statistic_refresh_log.sql
|
||
│ └── 2026_06_08_010_statistic_partitions_initial.sql
|
||
├── gateway/
|
||
│ ├── router/router.go # 修改：注册 7 路由
|
||
│ └── controller/statistic_controller.go # 新建：7 看板 controller
|
||
└── pkg/statistic/ # 新建：业务侧统一调用 SDK
|
||
└── client.go
|
||
```
|
||
|
||
### 修改
|
||
|
||
```
|
||
backend/
|
||
├── go.work # 修改：加 ./services/statisticService
|
||
├── services/socialService/service/asset_like_service.go # 修改：写 like_income_log 后调 TrackEvent
|
||
├── services/galleryService/service/exhibition_service.go # 修改：PlaceAsset / RemoveFromSlot 后调
|
||
├── services/taskService/service/revenue_service.go # 修改：OnExhibitionCompleted 后调
|
||
├── services/assetService/service/mint_service.go # 修改：CreateMintOrder 后调
|
||
├── services/assetService/service/asset_level_service.go # 修改：CheckUpgrade 后调
|
||
└── services/userService/service/user_service.go # 修改：UpdateCrystalBalance 后调
|
||
```
|
||
|
||
---
|
||
|
||
## 任务编排
|
||
|
||
**4 阶段 / 16 任务 / 每个任务包含 TDD 步骤**：
|
||
|
||
| 阶段 | 任务 | 周期（估）|
|
||
|------|------|----------|
|
||
| **P1 服务骨架** | T1-T3 | 2-3 天 |
|
||
| **P2 事件采集** | T4-T9 | 4-5 天 |
|
||
| **P3 看板 7 RPC** | T10-T14 | 4-5 天 |
|
||
| **P4 业务侧集成** | T15-T16 | 2-3 天 |
|
||
|
||
每个任务结束都应有：所有测试通过 + git commit + 关键验证勾选完成。
|
||
|
||
---
|
||
|
||
## P1 服务骨架
|
||
|
||
### Task 1: 项目初始化 + go.mod / go.work / 目录结构
|
||
|
||
**Files:**
|
||
- Create: `backend/services/statisticService/go.mod`
|
||
- Modify: `backend/go.work`
|
||
|
||
- [ ] **Step 1.1: 创建 go.mod**
|
||
|
||
```bash
|
||
cd backend/services/statisticService
|
||
go mod init github.com/topfans/backend/services/statisticService
|
||
```
|
||
|
||
- [ ] **Step 1.2: 添加核心依赖（沿用 taskService）**
|
||
|
||
```bash
|
||
go get github.com/topfans/backend/pkg/database@v0.0.0
|
||
go get github.com/topfans/backend/pkg/logger@v0.0.0
|
||
go get github.com/topfans/backend/pkg/health@v0.0.0
|
||
go get dubbo.apache.org/dubbo-go/v3
|
||
go get gorm.io/gorm
|
||
go get gorm.io/driver/postgres
|
||
go get github.com/redis/go-redis/v9
|
||
go get github.com/prometheus/client_golang/prometheus
|
||
```
|
||
|
||
> 实际版本号看 go.work 里 taskService 的 go.mod。复制其 `require` 块，删除 taskService 特定的包（assetLevelRepo 等）。
|
||
|
||
- [ ] **Step 1.3: 修改 go.work 加入新服务**
|
||
|
||
在 `backend/go.work` 的 `use` 列表里加：
|
||
```
|
||
./services/statisticService
|
||
```
|
||
|
||
- [ ] **Step 1.4: 创建子目录结构**
|
||
|
||
```bash
|
||
mkdir -p backend/services/statisticService/{config,model,repository,sink,service,provider,worker,handler,metrics,client,scripts/testhelper}
|
||
```
|
||
|
||
- [ ] **Step 1.5: 创建占位 main.go**
|
||
|
||
```go
|
||
// backend/services/statisticService/main.go
|
||
package main
|
||
|
||
import "fmt"
|
||
|
||
func main() {
|
||
fmt.Println("statisticService starting...")
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 1.6: 编译验证**
|
||
|
||
```bash
|
||
cd backend
|
||
go build ./services/statisticService
|
||
```
|
||
|
||
Expected: 编译成功，无输出。
|
||
|
||
- [ ] **Step 1.7: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/ backend/go.work
|
||
git commit -m "feat(statistic): scaffold statisticService with go.mod and directory structure"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 2: proto 定义（event.proto + statistic.proto）
|
||
|
||
**Files:**
|
||
- Create: `backend/proto/event.proto`
|
||
- Create: `backend/proto/statistic.proto`
|
||
|
||
> **关键：proto 冻结点。** P2/P3 业务逻辑依赖此定义。
|
||
|
||
- [ ] **Step 2.1: 创建 event.proto**
|
||
|
||
```bash
|
||
cat > backend/proto/event.proto << 'EOF'
|
||
syntax = "proto3";
|
||
package event;
|
||
option go_package = "github.com/topfans/backend/pkg/proto/event";
|
||
|
||
message Event {
|
||
string event_id = 1;
|
||
int64 user_id = 2;
|
||
int64 star_id = 3;
|
||
string event_type = 4;
|
||
int64 occurred_at = 5;
|
||
int64 received_at = 6;
|
||
map<string, string> properties = 7;
|
||
}
|
||
|
||
message BatchEventRequest {
|
||
repeated Event events = 1;
|
||
}
|
||
EOF
|
||
```
|
||
|
||
- [ ] **Step 2.2: 创建 statistic.proto（对齐 spec §2.3）**
|
||
|
||
```bash
|
||
cat > backend/proto/statistic.proto << 'EOF'
|
||
syntax = "proto3";
|
||
package statistic;
|
||
option go_package = "github.com/topfans/backend/pkg/proto/statistic";
|
||
import "event.proto";
|
||
|
||
service StatisticService {
|
||
rpc GetTodayOverview(GetTodayOverviewRequest) returns (GetTodayOverviewResponse);
|
||
rpc Get7DayIncomeCurve(Get7DayIncomeCurveRequest) returns (Get7DayIncomeCurveResponse);
|
||
rpc GetExhibitionIncomeSummary(GetExhibitionIncomeSummaryRequest) returns (GetExhibitionIncomeSummaryResponse);
|
||
rpc GetLikeIncomeByLevel(GetLikeIncomeByLevelRequest) returns (GetLikeIncomeByLevelResponse);
|
||
rpc GetTopAssetsByEarning(GetTopAssetsByEarningRequest) returns (GetTopAssetsByEarningResponse);
|
||
rpc GetAssetLevelDistribution(GetAssetLevelDistributionRequest) returns (GetAssetLevelDistributionResponse);
|
||
rpc GetAssetUpgradeProgress(GetAssetUpgradeProgressRequest) returns (GetAssetUpgradeProgressResponse);
|
||
rpc TrackEvent(event.Event) returns (TrackEventResponse);
|
||
rpc BatchTrackEvent(event.BatchEventRequest) returns (TrackEventResponse);
|
||
}
|
||
|
||
// ====== 1. 今日概览 ======
|
||
message GetTodayOverviewRequest { int64 star_id = 1; }
|
||
message GetTodayOverviewResponse {
|
||
int64 crystal_balance = 1;
|
||
int64 today_income = 2;
|
||
int32 week_rank = 3;
|
||
int32 week_total_users = 4;
|
||
}
|
||
|
||
// ====== 2. 七日收益曲线 ======
|
||
message Get7DayIncomeCurveRequest { int64 star_id = 1; }
|
||
message DailyIncomePoint {
|
||
string date = 1;
|
||
int64 income = 2;
|
||
bool is_today = 3;
|
||
bool is_peak = 4;
|
||
}
|
||
message Get7DayIncomeCurveResponse {
|
||
repeated DailyIncomePoint points = 1;
|
||
int64 total_income = 2;
|
||
int64 avg_income = 3;
|
||
}
|
||
|
||
// ====== 3. 展出收益中心 ======
|
||
message GetExhibitionIncomeSummaryRequest { int64 star_id = 1; }
|
||
message TopExhibitionItem {
|
||
int64 asset_id = 1;
|
||
string asset_name = 2;
|
||
string asset_thumb = 3;
|
||
string duration_7d = 4;
|
||
int64 earnings_7d = 5;
|
||
int32 avg_earnings = 6;
|
||
}
|
||
message GetExhibitionIncomeSummaryResponse {
|
||
int32 exhibiting_count = 1;
|
||
int32 starbook_count = 2;
|
||
string total_duration = 3;
|
||
int64 total_earnings = 4;
|
||
repeated TopExhibitionItem top5 = 5;
|
||
}
|
||
|
||
// ====== 4. 点赞收益按等级 ======
|
||
message GetLikeIncomeByLevelRequest { int64 star_id = 1; }
|
||
message LikeIncomeLevelItem {
|
||
string level = 1;
|
||
int32 asset_count = 2;
|
||
int64 total_income = 3;
|
||
string thumb = 4;
|
||
}
|
||
message GetLikeIncomeByLevelResponse {
|
||
int64 total_like_count = 1;
|
||
int64 total_income = 2;
|
||
repeated LikeIncomeLevelItem levels = 3;
|
||
}
|
||
|
||
// ====== 5. 藏品 TOP5 ======
|
||
message GetTopAssetsByEarningRequest { int64 star_id = 1; }
|
||
message TopAssetItem {
|
||
int64 asset_id = 1;
|
||
string asset_name = 2;
|
||
string asset_thumb = 3;
|
||
int64 total_earnings = 4;
|
||
int32 rank = 5;
|
||
}
|
||
message GetTopAssetsByEarningResponse {
|
||
repeated TopAssetItem items = 1;
|
||
}
|
||
|
||
// ====== 6. 藏品等级分布 ======
|
||
message GetAssetLevelDistributionRequest { int64 star_id = 1; }
|
||
message AssetLevelItem {
|
||
string level = 1;
|
||
int32 count = 2;
|
||
int32 total = 3;
|
||
}
|
||
message GetAssetLevelDistributionResponse {
|
||
repeated AssetLevelItem items = 1;
|
||
}
|
||
|
||
// ====== 7. 升级进度 ======
|
||
message GetAssetUpgradeProgressRequest { int64 star_id = 1; }
|
||
message UpcomingLevelUpItem {
|
||
int64 asset_id = 1;
|
||
string asset_name = 2;
|
||
string asset_thumb = 3;
|
||
int32 like_progress = 4;
|
||
int32 duration_progress = 5;
|
||
}
|
||
message RecentLevelUpItem {
|
||
int64 asset_id = 1;
|
||
string asset_name = 2;
|
||
string asset_thumb = 3;
|
||
string new_level = 4;
|
||
int64 upgrade_time = 5;
|
||
}
|
||
message GetAssetUpgradeProgressResponse {
|
||
repeated UpcomingLevelUpItem upcoming = 1;
|
||
repeated RecentLevelUpItem recent = 2;
|
||
}
|
||
|
||
// ====== 事件采集响应 ======
|
||
message TrackEventResponse {
|
||
int32 accepted = 1;
|
||
int32 rejected = 2;
|
||
}
|
||
EOF
|
||
```
|
||
|
||
- [ ] **Step 2.3: 编译 proto（生成 Go 代码）**
|
||
|
||
```bash
|
||
cd backend
|
||
# 实际项目用 make / 脚本生成，看 Makefile 或 scripts/ 里 protoc 命令
|
||
ls scripts/ 2>/dev/null
|
||
# 或用标准命令：
|
||
protoc --go_out=pkg/proto --go_triple_out=pkg/proto \
|
||
-I proto proto/event.proto proto/statistic.proto
|
||
```
|
||
|
||
Expected: `pkg/proto/event/` 和 `pkg/proto/statistic/` 下生成 `.pb.go` 和 `.triple.go`。
|
||
|
||
- [ ] **Step 2.4: 验证编译**
|
||
|
||
```bash
|
||
go build ./pkg/proto/event/ ./pkg/proto/statistic/
|
||
```
|
||
|
||
Expected: 编译成功。
|
||
|
||
- [ ] **Step 2.5: Commit**
|
||
|
||
```bash
|
||
git add backend/proto/ backend/pkg/proto/event/ backend/pkg/proto/statistic/
|
||
git commit -m "feat(statistic): add event.proto and statistic.proto with 7 RPCs + 2 event RPCs"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 3: config + main.go 启动 + healthz + metrics 骨架 + 10 SQL 迁移
|
||
|
||
**Files:**
|
||
- Create: `backend/services/statisticService/config/statistic_config.go`
|
||
- Modify: `backend/services/statisticService/main.go`
|
||
- Create: `backend/services/statisticService/handler/healthz.go`
|
||
- Create: `backend/services/statisticService/metrics/metrics.go`
|
||
- Create: 10 SQL migration files under `backend/migrations/`
|
||
|
||
- [ ] **Step 3.1: 创建 config（沿用 taskService 模式）**
|
||
|
||
```go
|
||
// backend/services/statisticService/config/statistic_config.go
|
||
package config
|
||
|
||
import (
|
||
"flag"
|
||
"log"
|
||
"os"
|
||
"strconv"
|
||
"time"
|
||
)
|
||
|
||
type DatabaseConfig struct {
|
||
Host, Password, DBName, SSLMode, TimeZone string
|
||
Port int
|
||
User string
|
||
Schema string
|
||
}
|
||
|
||
type RedisConfig struct {
|
||
URL string
|
||
}
|
||
|
||
type RefreshIntervals struct {
|
||
DailyUserIncome time.Duration
|
||
DailyExhibitionRevenue time.Duration
|
||
DailyLikeIncome time.Duration
|
||
AssetLevelDistribution time.Duration
|
||
WeeklyUserIncome time.Duration
|
||
UpcomingLevelUps time.Duration
|
||
}
|
||
|
||
type ChannelConfig struct {
|
||
EventChannelCapacity int
|
||
EventWorkerCount int
|
||
EventBatchSize int
|
||
EventBatchInterval time.Duration
|
||
}
|
||
|
||
type PartitionConfig struct {
|
||
RetentionDays int
|
||
PreCreateDays int
|
||
}
|
||
|
||
type ExtensionConfig struct {
|
||
EnableOLAPDualWrite bool
|
||
EnableRealtimeChannel bool
|
||
EnableSDKEndpoint bool
|
||
EnableSampling bool
|
||
}
|
||
|
||
var (
|
||
DBConfig = &DatabaseConfig{Schema: "statistic"}
|
||
RedisCfg = &RedisConfig{URL: "redis://localhost:6379/0"}
|
||
RefreshCfg = &RefreshIntervals{
|
||
DailyUserIncome: 5 * time.Minute,
|
||
DailyExhibitionRevenue: 5 * time.Minute,
|
||
DailyLikeIncome: 5 * time.Minute,
|
||
AssetLevelDistribution: 15 * time.Minute,
|
||
WeeklyUserIncome: 5 * time.Minute,
|
||
UpcomingLevelUps: 15 * time.Minute,
|
||
}
|
||
ChannelCfg = &ChannelConfig{
|
||
EventChannelCapacity: 1000,
|
||
EventWorkerCount: 1,
|
||
EventBatchSize: 100,
|
||
EventBatchInterval: 1 * time.Second,
|
||
}
|
||
PartitionCfg = &PartitionConfig{RetentionDays: 30, PreCreateDays: 7}
|
||
ExtCfg = &ExtensionConfig{ // 全部默认 false
|
||
EnableOLAPDualWrite: false,
|
||
EnableRealtimeChannel: false,
|
||
EnableSDKEndpoint: false,
|
||
EnableSampling: false,
|
||
}
|
||
)
|
||
|
||
func getEnv(key, fallback string) string {
|
||
if v := os.Getenv(key); v != "" { return v }
|
||
return fallback
|
||
}
|
||
|
||
func getEnvInt(key string, fallback int) int {
|
||
if v := os.Getenv(key); v != "" {
|
||
if n, err := strconv.Atoi(v); err == nil { return n }
|
||
}
|
||
return fallback
|
||
}
|
||
|
||
func getEnvDuration(key string, fallback time.Duration) time.Duration {
|
||
if v := os.Getenv(key); v != "" {
|
||
if d, err := time.ParseDuration(v); err == nil { return d }
|
||
}
|
||
return fallback
|
||
}
|
||
|
||
func getEnvBool(key string, fallback bool) bool {
|
||
if v := os.Getenv(key); v != "" {
|
||
if b, err := strconv.ParseBool(v); err == nil { return b }
|
||
}
|
||
return fallback
|
||
}
|
||
|
||
func InitConfig() {
|
||
flag.StringVar(&DBConfig.Host, "db-host", getEnv("STATISTIC_DB_HOST", "localhost"), "")
|
||
flag.IntVar(&DBConfig.Port, "db-port", getEnvInt("STATISTIC_DB_PORT", 5432), "")
|
||
flag.StringVar(&DBConfig.User, "db-user", getEnv("STATISTIC_DB_USER", "postgres"), "")
|
||
flag.StringVar(&DBConfig.Password, "db-password", getEnv("STATISTIC_DB_PASSWORD", ""), "")
|
||
flag.StringVar(&DBConfig.DBName, "db-name", getEnv("STATISTIC_DB_NAME", "topfans"), "")
|
||
flag.StringVar(&DBConfig.SSLMode, "db-sslmode", "disable", "")
|
||
flag.StringVar(&DBConfig.Schema, "db-schema", getEnv("STATISTIC_DB_SCHEMA", "statistic"), "")
|
||
|
||
flag.StringVar(&RedisCfg.URL, "redis-url", getEnv("STATISTIC_REDIS_URL", "redis://localhost:6379/0"), "")
|
||
|
||
flag.IntVar(&ChannelCfg.EventChannelCapacity, "event-channel-capacity", getEnvInt("STATISTIC_EVENT_CHANNEL_CAPACITY", 1000), "")
|
||
flag.IntVar(&ChannelCfg.EventBatchSize, "event-batch-size", getEnvInt("STATISTIC_EVENT_BATCH_SIZE", 100), "")
|
||
flag.DurationVar(&ChannelCfg.EventBatchInterval, "event-batch-interval", getEnvDuration("STATISTIC_EVENT_BATCH_INTERVAL", time.Second), "")
|
||
|
||
flag.IntVar(&PartitionCfg.RetentionDays, "partition-retention-days", getEnvInt("STATISTIC_PARTITION_RETENTION_DAYS", 30), "")
|
||
flag.IntVar(&PartitionCfg.PreCreateDays, "partition-precreate-days", getEnvInt("STATISTIC_PARTITION_PRECREATE_DAYS", 7), "")
|
||
|
||
flag.BoolVar(&ExtCfg.EnableOLAPDualWrite, "enable-olap", getEnvBool("STATISTIC_ENABLE_OLAP_DUAL_WRITE", false), "")
|
||
flag.BoolVar(&ExtCfg.EnableRealtimeChannel, "enable-realtime", getEnvBool("STATISTIC_ENABLE_REALTIME_CHANNEL", false), "")
|
||
flag.BoolVar(&ExtCfg.EnableSDKEndpoint, "enable-sdk", getEnvBool("STATISTIC_ENABLE_SDK_ENDPOINT", false), "")
|
||
flag.BoolVar(&ExtCfg.EnableSampling, "enable-sampling", getEnvBool("STATISTIC_ENABLE_SAMPLING", false), "")
|
||
|
||
flag.Parse()
|
||
log.Println("statisticService 配置初始化完成")
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 3.2: 创建 metrics 骨架（声明所有指标，数据后续埋）**
|
||
|
||
```go
|
||
// backend/services/statisticService/metrics/metrics.go
|
||
package metrics
|
||
|
||
import (
|
||
"github.com/prometheus/client_golang/prometheus"
|
||
"github.com/prometheus/client_golang/prometheus/promauto"
|
||
)
|
||
|
||
// 看板 RPC
|
||
var DashboardRPCTotal = promauto.NewCounterVec(prometheus.CounterOpts{
|
||
Name: "dashboard_rpc_total", Help: "Dashboard RPC total",
|
||
}, []string{"rpc", "status"})
|
||
|
||
var DashboardRPCDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
|
||
Name: "dashboard_rpc_duration_seconds", Help: "Dashboard RPC duration",
|
||
}, []string{"rpc"})
|
||
|
||
var DashboardCacheHitRate = promauto.NewGauge(prometheus.GaugeOpts{
|
||
Name: "dashboard_cache_hit_rate", Help: "Dashboard cache hit rate",
|
||
})
|
||
|
||
// 事件采集
|
||
var EventTrackTotal = promauto.NewCounterVec(prometheus.CounterOpts{
|
||
Name: "event_track_total", Help: "Event track total",
|
||
}, []string{"event_type", "result"})
|
||
|
||
var EventChannelSize = promauto.NewGauge(prometheus.GaugeOpts{
|
||
Name: "event_channel_size", Help: "Event channel current size",
|
||
})
|
||
|
||
var EventChannelCapacity = promauto.NewGauge(prometheus.GaugeOpts{
|
||
Name: "event_channel_capacity", Help: "Event channel capacity",
|
||
})
|
||
|
||
var EventDroppedTotal = promauto.NewCounter(prometheus.CounterOpts{
|
||
Name: "event_dropped_total", Help: "Event dropped total",
|
||
})
|
||
|
||
var EventDBInsertTotal = promauto.NewCounterVec(prometheus.CounterOpts{
|
||
Name: "event_db_insert_total", Help: "Event DB insert total",
|
||
}, []string{"status"})
|
||
|
||
// 物化视图
|
||
var MVRefreshTotal = promauto.NewCounterVec(prometheus.CounterOpts{
|
||
Name: "mv_refresh_total", Help: "Materialized view refresh total",
|
||
}, []string{"mv_name", "status"})
|
||
|
||
var MVRefreshDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
|
||
Name: "mv_refresh_duration_seconds", Help: "MV refresh duration",
|
||
}, []string{"mv_name"})
|
||
|
||
// Worker
|
||
var WorkerRunningCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
|
||
Name: "worker_running_count", Help: "Worker running count",
|
||
}, []string{"worker_name"})
|
||
|
||
// 分区
|
||
var EventsPartitionCount = promauto.NewGauge(prometheus.GaugeOpts{
|
||
Name: "events_partition_count", Help: "Events partition count",
|
||
})
|
||
```
|
||
|
||
- [ ] **Step 3.3: 创建 healthz**
|
||
|
||
```go
|
||
// backend/services/statisticService/handler/healthz.go
|
||
package handler
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"net/http"
|
||
"time"
|
||
|
||
"github.com/gin-gonic/gin"
|
||
"github.com/redis/go-redis/v9"
|
||
"github.com/topfans/backend/services/statisticService/metrics"
|
||
)
|
||
|
||
type Healthz struct {
|
||
db *sql.DB
|
||
redis *redis.Client
|
||
}
|
||
|
||
func NewHealthz(db *sql.DB, redis *redis.Client) *Healthz {
|
||
return &Healthz{db: db, redis: redis}
|
||
}
|
||
|
||
func (h *Healthz) Register(r *gin.Engine, port int) {
|
||
r.GET("/metrics", gin.WrapH(metricsHandler())) // 暴露 Prometheus
|
||
r.GET("/healthz", func(c *gin.Context) {
|
||
ctx, cancel := context.WithTimeout(c.Request.Context(), 1*time.Second)
|
||
defer cancel()
|
||
|
||
status := gin.H{"status": "ok"}
|
||
if err := h.db.PingContext(ctx); err != nil {
|
||
status["db"] = "down"
|
||
} else {
|
||
status["db"] = "up"
|
||
}
|
||
if err := h.redis.Ping(ctx).Err(); err != nil {
|
||
status["redis"] = "down"
|
||
} else {
|
||
status["redis"] = "up"
|
||
}
|
||
c.JSON(http.StatusOK, status)
|
||
})
|
||
}
|
||
```
|
||
|
||
> **实际项目**用 `pkg/health` 组件，参考 taskService main.go:131-133。`health.NewHandler("statistic-service", healthPort).Start()`。本计划用自实现版，最终应替换为项目既有组件。
|
||
|
||
- [ ] **Step 3.4: 创建 main.go（启动骨架，不含业务）**
|
||
|
||
```go
|
||
// backend/services/statisticService/main.go
|
||
package main
|
||
|
||
import (
|
||
"database/sql"
|
||
"flag"
|
||
"fmt"
|
||
"os"
|
||
"os/signal"
|
||
"syscall"
|
||
|
||
_ "github.com/lib/pq"
|
||
"github.com/redis/go-redis/v9"
|
||
"github.com/topfans/backend/pkg/logger"
|
||
"github.com/topfans/backend/services/statisticService/config"
|
||
"github.com/topfans/backend/services/statisticService/handler"
|
||
)
|
||
|
||
var port = flag.Int("port", 20009, "Dubbo service port")
|
||
|
||
func main() {
|
||
env := os.Getenv("ENV")
|
||
if env == "" { env = "development" }
|
||
if err := logger.Init(logger.Config{ServiceName: "statistic-service", Environment: env}); err != nil {
|
||
panic(fmt.Sprintf("Failed to init logger: %v", err))
|
||
}
|
||
defer logger.Sync()
|
||
|
||
config.InitConfig()
|
||
|
||
// DB
|
||
db, err := sql.Open("postgres", fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=%s",
|
||
config.DBConfig.Host, config.DBConfig.Port, config.DBConfig.User, config.DBConfig.Password,
|
||
config.DBConfig.DBName, config.DBConfig.SSLMode))
|
||
if err != nil { logger.Logger.Fatal(fmt.Sprintf("Failed to open DB: %v", err)) }
|
||
db.SetMaxOpenConns(50)
|
||
if err := db.Ping(); err != nil { logger.Logger.Fatal(fmt.Sprintf("DB ping: %v", err)) }
|
||
defer db.Close()
|
||
|
||
// Redis
|
||
opt, _ := redis.ParseURL(config.RedisCfg.URL)
|
||
rdb := redis.NewClient(opt)
|
||
defer rdb.Close()
|
||
|
||
// Healthz HTTP server on port+1000=21009
|
||
healthPort := *port + 1000
|
||
h := handler.NewHealthz(db, rdb)
|
||
go h.Start(healthPort)
|
||
|
||
logger.Logger.Info(fmt.Sprintf("statisticService started, dubbo=%d healthz=%d", *port, healthPort))
|
||
|
||
// Graceful shutdown
|
||
quit := make(chan os.Signal, 1)
|
||
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
|
||
<-quit
|
||
}
|
||
```
|
||
|
||
并在 healthz.go 加 `Start` 方法：
|
||
|
||
```go
|
||
// 加到 backend/services/statisticService/handler/healthz.go
|
||
func (h *Healthz) Start(port int) {
|
||
r := gin.Default()
|
||
h.Register(r, port)
|
||
go func() {
|
||
if err := r.Run(fmt.Sprintf(":%d", port)); err != nil {
|
||
logger.Logger.Fatal(...)
|
||
}
|
||
}()
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 3.5: 创建 10 个 SQL 迁移文件（参考 spec §3.2-3.5）**
|
||
|
||
每个文件包含一个 DDL + 必要的索引。**关键 SQL 摘要：**
|
||
|
||
`backend/migrations/2026_06_08_001_statistic_events.sql`：
|
||
|
||
```sql
|
||
CREATE SCHEMA IF NOT EXISTS statistic;
|
||
|
||
CREATE TABLE IF NOT EXISTS statistic.events (
|
||
id BIGSERIAL,
|
||
event_id UUID NOT NULL,
|
||
user_id BIGINT NOT NULL,
|
||
star_id BIGINT NOT NULL,
|
||
event_type VARCHAR(64) NOT NULL,
|
||
occurred_at TIMESTAMPTZ NOT NULL,
|
||
received_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||
properties JSONB NOT NULL DEFAULT '{}',
|
||
PRIMARY KEY (id, received_at)
|
||
) PARTITION BY RANGE (received_at);
|
||
|
||
CREATE UNIQUE INDEX IF NOT EXISTS idx_events_event_id ON statistic.events (event_id, received_at);
|
||
CREATE INDEX IF NOT EXISTS idx_events_user_star_type_time
|
||
ON statistic.events (user_id, star_id, event_type, received_at DESC);
|
||
CREATE INDEX IF NOT EXISTS idx_events_star_type_time
|
||
ON statistic.events (star_id, event_type, received_at DESC);
|
||
CREATE INDEX IF NOT EXISTS idx_events_properties_gin ON statistic.events USING GIN (properties);
|
||
|
||
-- 初始 7 天分区（从今天起）
|
||
DO $$
|
||
DECLARE
|
||
i INT;
|
||
d DATE;
|
||
n DATE;
|
||
BEGIN
|
||
FOR i IN 0..6 LOOP
|
||
d := CURRENT_DATE + i;
|
||
n := d + 1;
|
||
EXECUTE format(
|
||
'CREATE TABLE IF NOT EXISTS statistic.events_%s PARTITION OF statistic.events FOR VALUES FROM (%L) TO (%L)',
|
||
to_char(d, 'YYYY_MM_DD'), d::text, n::text
|
||
);
|
||
END LOOP;
|
||
END $$;
|
||
```
|
||
|
||
`2026_06_08_002_statistic_mv_daily_user_income.sql` - 见 spec §3.4 MV1 SQL
|
||
|
||
`2026_06_08_003_statistic_mv_daily_exhibition_revenue.sql` - 见 spec §3.4 MV2 SQL
|
||
|
||
`2026_06_08_004_statistic_mv_daily_like_income.sql` - 见 spec §3.4 MV3 SQL
|
||
|
||
`2026_06_08_005_statistic_mv_asset_level_distribution.sql` - 见 spec §3.4 MV4 SQL
|
||
|
||
`2026_06_08_006_statistic_metric_weekly_user_income.sql` - 见 spec §3.5 weekly 表
|
||
|
||
`2026_06_08_007_statistic_metric_recent_level_ups.sql` - 见 spec §3.5 recent_level_ups 表
|
||
|
||
`2026_06_08_008_statistic_metric_upcoming_level_ups.sql` - 见 spec §3.5 upcoming_level_ups 表
|
||
|
||
`2026_06_08_009_statistic_refresh_log.sql`：
|
||
|
||
```sql
|
||
CREATE TABLE IF NOT EXISTS statistic.refresh_log (
|
||
id BIGSERIAL PRIMARY KEY,
|
||
mv_name VARCHAR(128) NOT NULL,
|
||
started_at TIMESTAMPTZ NOT NULL,
|
||
finished_at TIMESTAMPTZ,
|
||
row_count BIGINT,
|
||
status VARCHAR(16) NOT NULL,
|
||
error_message TEXT
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_refresh_log_mv_time
|
||
ON statistic.refresh_log (mv_name, started_at DESC);
|
||
```
|
||
|
||
- [ ] **Step 3.6: 跑 DB 迁移**
|
||
|
||
```bash
|
||
cd backend
|
||
# 用项目现有的迁移工具（看 scripts/）
|
||
ls scripts/ 2>/dev/null
|
||
# 或手动 psql：
|
||
psql -h localhost -U postgres -d topfans -f migrations/2026_06_08_001_statistic_events.sql
|
||
# ... 10 个文件依次执行
|
||
```
|
||
|
||
Expected: 10 个 SQL 全部成功，\dt statistic.* 看到 10+ 表。
|
||
|
||
- [ ] **Step 3.7: 编译 + 运行**
|
||
|
||
```bash
|
||
go build ./services/statisticService
|
||
./services/statisticService/bin/statisticService
|
||
```
|
||
|
||
Expected: 启动日志显示 port 20009 + healthz 21009。
|
||
|
||
- [ ] **Step 3.8: 验证 healthz**
|
||
|
||
```bash
|
||
curl http://localhost:21009/healthz
|
||
```
|
||
|
||
Expected: `{"db":"up","redis":"up","status":"ok"}`
|
||
|
||
- [ ] **Step 3.9: 跑 P1 末预检查清单**
|
||
|
||
> 必做！spec §0.1.3 P1 末预检查清单 10 项。
|
||
|
||
```bash
|
||
# 1. socialService
|
||
grep -r "like_income_log" backend/services/socialService/
|
||
# 2. galleryService
|
||
grep -rn "^func.*PlaceAsset\|^func.*RemoveFromSlot" backend/services/galleryService/service/
|
||
# 3. taskService
|
||
grep -n "OnExhibitionCompleted" backend/services/taskService/service/revenue_service.go
|
||
# 4. assetService
|
||
grep -n "CreateMintOrder\|CheckUpgrade\|logLevelChange" backend/services/assetService/service/{mint_service,asset_level_service}.go
|
||
# 5. userService
|
||
grep -n "UpdateCrystalBalance" backend/services/userService/service/user_service.go
|
||
# 6. public.assets 表结构
|
||
psql -c "\d+ public.assets"
|
||
# 7. 公共日志表
|
||
psql -c "\dt public.*log*"
|
||
# 8. 端口
|
||
lsof -i :20009
|
||
# 9. 业务服务可达
|
||
nc -zv localhost 20000; nc -zv localhost 20001; nc -zv localhost 20002; nc -zv localhost 20003; nc -zv localhost 20006
|
||
# 10. schema 权限
|
||
psql -U postgres -c "CREATE SCHEMA statistic_test_perm; DROP SCHEMA statistic_test_perm;"
|
||
```
|
||
|
||
Expected: 全部通过。
|
||
|
||
- [ ] **Step 3.10: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/ backend/migrations/
|
||
git commit -m "feat(statistic): P1 scaffolding - config, main, healthz, metrics, 10 SQL migrations"
|
||
```
|
||
|
||
---
|
||
|
||
## P2 事件采集
|
||
|
||
### Task 4: Event 模型 + EventSink 接口 + ChannelEventSink
|
||
|
||
**Files:**
|
||
- Create: `backend/services/statisticService/model/event.go`
|
||
- Create: `backend/services/statisticService/sink/event_sink.go`
|
||
- Create: `backend/services/statisticService/sink/channel_sink.go`
|
||
- Create: `backend/services/statisticService/sink/channel_sink_test.go`
|
||
|
||
- [ ] **Step 4.1: 写 Event 模型（写测试先）**
|
||
|
||
`backend/services/statisticService/model/event_test.go`：
|
||
|
||
```go
|
||
package model
|
||
|
||
import (
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func TestEvent_ToJSON(t *testing.T) {
|
||
e := &Event{
|
||
EventID: "uuid-123",
|
||
UserID: 100,
|
||
StarID: 1,
|
||
EventType: "asset.like",
|
||
OccurredAt: time.Unix(1700000000, 0),
|
||
ReceivedAt: time.Unix(1700000001, 0),
|
||
Properties: map[string]string{"asset_id": "456"},
|
||
}
|
||
j := e.ToJSON()
|
||
if j["event_id"] != "uuid-123" { t.Fatal("event_id mismatch") }
|
||
if j["properties"].(map[string]interface{})["asset_id"] != "456" { t.Fatal("properties mismatch") }
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 4.2: 跑测试（应失败）**
|
||
|
||
```bash
|
||
cd backend/services/statisticService
|
||
go test ./model/ -v
|
||
```
|
||
|
||
Expected: FAIL - "undefined: Event"
|
||
|
||
- [ ] **Step 4.3: 实现 Event**
|
||
|
||
`backend/services/statisticService/model/event.go`：
|
||
|
||
```go
|
||
package model
|
||
|
||
import "time"
|
||
|
||
type Event struct {
|
||
EventID string `json:"event_id"`
|
||
UserID int64 `json:"user_id"`
|
||
StarID int64 `json:"star_id"`
|
||
EventType string `json:"event_type"`
|
||
OccurredAt time.Time `json:"occurred_at"`
|
||
ReceivedAt time.Time `json:"received_at"`
|
||
Properties map[string]string `json:"properties"`
|
||
}
|
||
|
||
func (e *Event) ToJSON() map[string]interface{} {
|
||
return map[string]interface{}{
|
||
"event_id": e.EventID,
|
||
"user_id": e.UserID,
|
||
"star_id": e.StarID,
|
||
"event_type": e.EventType,
|
||
"occurred_at": e.OccurredAt.UnixMilli(),
|
||
"received_at": e.ReceivedAt.UnixMilli(),
|
||
"properties": e.Properties,
|
||
}
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 4.4: 跑测试（应通过）**
|
||
|
||
```bash
|
||
go test ./model/ -v
|
||
```
|
||
|
||
Expected: PASS
|
||
|
||
- [ ] **Step 4.5: 写 EventSink 接口 + ChannelEventSink 测试**
|
||
|
||
`backend/services/statisticService/sink/channel_sink_test.go`：
|
||
|
||
```go
|
||
package sink
|
||
|
||
import (
|
||
"context"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func TestChannelEventSink_Submit(t *testing.T) {
|
||
ch := make(chan *model.Event, 10)
|
||
s := NewChannelEventSink(ch)
|
||
e := &model.Event{EventID: "test-1"}
|
||
if err := s.Submit(context.Background(), e); err != nil {
|
||
t.Fatalf("Submit failed: %v", err)
|
||
}
|
||
select {
|
||
case got := <-ch:
|
||
if got.EventID != "test-1" { t.Fatal("event mismatch") }
|
||
case <-time.After(100 * time.Millisecond):
|
||
t.Fatal("no event received")
|
||
}
|
||
}
|
||
|
||
func TestChannelEventSink_SubmitBatch(t *testing.T) {
|
||
ch := make(chan *model.Event, 10)
|
||
s := NewChannelEventSink(ch)
|
||
events := []*model.Event{{EventID: "a"}, {EventID: "b"}}
|
||
if err := s.SubmitBatch(context.Background(), events); err != nil {
|
||
t.Fatalf("SubmitBatch failed: %v", err)
|
||
}
|
||
if len(ch) != 2 { t.Fatalf("expected 2 events, got %d", len(ch)) }
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 4.6: 跑测试（应失败）**
|
||
|
||
```bash
|
||
go test ./sink/ -v
|
||
```
|
||
|
||
Expected: FAIL
|
||
|
||
- [ ] **Step 4.7: 实现接口和 ChannelEventSink**
|
||
|
||
`backend/services/statisticService/sink/event_sink.go`：
|
||
|
||
```go
|
||
package sink
|
||
|
||
import (
|
||
"context"
|
||
"github.com/topfans/backend/services/statisticService/model"
|
||
)
|
||
|
||
type EventSink interface {
|
||
Submit(ctx context.Context, e *model.Event) error
|
||
SubmitBatch(ctx context.Context, es []*model.Event) error
|
||
Close() error
|
||
}
|
||
```
|
||
|
||
`backend/services/statisticService/sink/channel_sink.go`：
|
||
|
||
```go
|
||
package sink
|
||
|
||
import (
|
||
"context"
|
||
"github.com/topfans/backend/services/statisticService/model"
|
||
)
|
||
|
||
type ChannelEventSink struct {
|
||
ch chan<- *model.Event
|
||
}
|
||
|
||
func NewChannelEventSink(ch chan<- *model.Event) *ChannelEventSink {
|
||
return &ChannelEventSink{ch: ch}
|
||
}
|
||
|
||
func (s *ChannelEventSink) Submit(ctx context.Context, e *model.Event) error {
|
||
select {
|
||
case s.ch <- e:
|
||
return nil
|
||
default:
|
||
return ErrChannelFull
|
||
}
|
||
}
|
||
|
||
func (s *ChannelEventSink) SubmitBatch(ctx context.Context, es []*model.Event) error {
|
||
for _, e := range es {
|
||
if err := s.Submit(ctx, e); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *ChannelEventSink) Close() error { return nil }
|
||
|
||
var ErrChannelFull = errors.New("event channel full")
|
||
```
|
||
|
||
补 `import "errors"`。
|
||
|
||
- [ ] **Step 4.8: 跑测试（应通过）**
|
||
|
||
```bash
|
||
go test ./sink/ -v
|
||
```
|
||
|
||
Expected: PASS
|
||
|
||
- [ ] **Step 4.9: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/model/ backend/services/statisticService/sink/
|
||
git commit -m "feat(statistic): Event model, EventSink interface, ChannelEventSink (P2 T4)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 5: event_repo + event_service（事件校验 + 落库）
|
||
|
||
**Files:**
|
||
- Create: `backend/services/statisticService/repository/event_repo.go`
|
||
- Create: `backend/services/statisticService/service/event_service.go`
|
||
- Tests
|
||
|
||
- [ ] **Step 5.1: 写 event_repo 批量插入测试**
|
||
|
||
`backend/services/statisticService/repository/event_repo_test.go`：
|
||
|
||
```go
|
||
package repository
|
||
|
||
import (
|
||
"context"
|
||
"os"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func setupTestDB(t *testing.T) (*sql.DB, func()) {
|
||
dsn := os.Getenv("TEST_DATABASE_URL")
|
||
if dsn == "" {
|
||
t.Skip("TEST_DATABASE_URL not set, skipping integration test")
|
||
}
|
||
db, err := sql.Open("postgres", dsn)
|
||
if err != nil { t.Fatal(err) }
|
||
// 用 schema=test, 每次 truncate events
|
||
if _, err := db.Exec("CREATE SCHEMA IF NOT EXISTS statistic_test"); err != nil { t.Fatal(err) }
|
||
return db, func() {
|
||
db.Exec("TRUNCATE TABLE statistic_test.events")
|
||
db.Close()
|
||
}
|
||
}
|
||
|
||
func TestEventRepo_InsertBatch(t *testing.T) {
|
||
db, cleanup := setupTestDB(t)
|
||
defer cleanup()
|
||
repo := NewEventRepository(db, "statistic_test")
|
||
events := []*model.Event{
|
||
{EventID: "uuid-1", UserID: 100, StarID: 1, EventType: "asset.like",
|
||
OccurredAt: time.Now(), ReceivedAt: time.Now(), Properties: map[string]string{"asset_id": "456"}},
|
||
}
|
||
inserted, err := repo.InsertBatch(context.Background(), events)
|
||
if err != nil { t.Fatal(err) }
|
||
if inserted != 1 { t.Fatalf("expected 1, got %d", inserted) }
|
||
}
|
||
|
||
func TestEventRepo_Dedup(t *testing.T) {
|
||
db, cleanup := setupTestDB(t)
|
||
defer cleanup()
|
||
repo := NewEventRepository(db, "statistic_test")
|
||
e := &model.Event{EventID: "uuid-dup", UserID: 100, StarID: 1, EventType: "asset.like",
|
||
OccurredAt: time.Now(), ReceivedAt: time.Now(), Properties: map[string]string{}}
|
||
repo.InsertBatch(context.Background(), []*model.Event{e})
|
||
inserted, _ := repo.InsertBatch(context.Background(), []*model.Event{e}) // 重复
|
||
if inserted != 0 { t.Fatalf("expected 0 (dedup), got %d", inserted) }
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 5.2: 跑测试（应失败）**
|
||
|
||
```bash
|
||
TEST_DATABASE_URL="postgres://postgres:postgres@localhost:5432/topfans_test?sslmode=disable" \
|
||
go test ./repository/ -v -run TestEventRepo
|
||
```
|
||
|
||
Expected: FAIL
|
||
|
||
- [ ] **Step 5.3: 实现 event_repo**
|
||
|
||
`backend/services/statisticService/repository/event_repo.go`：
|
||
|
||
```go
|
||
package repository
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
|
||
"github.com/topfans/backend/services/statisticService/model"
|
||
)
|
||
|
||
type EventRepository struct {
|
||
db *sql.DB
|
||
schema string
|
||
}
|
||
|
||
func NewEventRepository(db *sql.DB, schema string) *EventRepository {
|
||
return &EventRepository{db: db, schema: schema}
|
||
}
|
||
|
||
func (r *EventRepository) InsertBatch(ctx context.Context, events []*model.Event) (int, error) {
|
||
if len(events) == 0 { return 0, nil }
|
||
|
||
placeholders := make([]string, 0, len(events))
|
||
args := make([]interface{}, 0, len(events)*7)
|
||
for _, e := range events {
|
||
props, _ := json.Marshal(e.Properties)
|
||
placeholders = append(placeholders, fmt.Sprintf("($%d, $%d, $%d, $%d, $%d, $%d, $%d)",
|
||
len(args)+1, len(args)+2, len(args)+3, len(args)+4, len(args)+5, len(args)+6, len(args)+7))
|
||
args = append(args, e.EventID, e.UserID, e.StarID, e.EventType, e.OccurredAt, e.ReceivedAt, string(props))
|
||
}
|
||
|
||
query := fmt.Sprintf(`
|
||
INSERT INTO %s.events (event_id, user_id, star_id, event_type, occurred_at, received_at, properties)
|
||
VALUES %s
|
||
ON CONFLICT (event_id, received_at) DO NOTHING
|
||
`, r.schema, strings.Join(placeholders, ","))
|
||
|
||
res, err := r.db.ExecContext(ctx, query, args...)
|
||
if err != nil { return 0, err }
|
||
n, _ := res.RowsAffected()
|
||
return int(n), nil
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 5.4: 跑测试（应通过）**
|
||
|
||
```bash
|
||
TEST_DATABASE_URL="postgres://postgres:postgres@localhost:5432/topfans_test?sslmode=disable" \
|
||
go test ./repository/ -v -run TestEventRepo
|
||
```
|
||
|
||
Expected: PASS
|
||
|
||
- [ ] **Step 5.5: 写 event_service 校验测试**
|
||
|
||
`backend/services/statisticService/service/event_service_test.go`：
|
||
|
||
```go
|
||
package service
|
||
|
||
import (
|
||
"context"
|
||
"testing"
|
||
"github.com/topfans/backend/services/statisticService/model"
|
||
"github.com/topfans/backend/services/statisticService/sink"
|
||
)
|
||
|
||
type mockSink struct{ events []*model.Event }
|
||
func (m *mockSink) Submit(ctx context.Context, e *model.Event) error {
|
||
m.events = append(m.events, e); return nil
|
||
}
|
||
func (m *mockSink) SubmitBatch(ctx context.Context, es []*model.Event) error {
|
||
m.events = append(m.events, es...); return nil
|
||
}
|
||
func (m *mockSink) Close() error { return nil }
|
||
|
||
func TestEventService_TrackEvent_Success(t *testing.T) {
|
||
ms := &mockSink{}
|
||
svc := NewEventService(ms, []string{"asset.like", "asset.mint"})
|
||
resp, err := svc.TrackEvent(context.Background(), &model.Event{
|
||
EventID: "u1", UserID: 1, StarID: 1, EventType: "asset.like",
|
||
OccurredAt: time.Now(), ReceivedAt: time.Now(), Properties: map[string]string{},
|
||
})
|
||
if err != nil { t.Fatal(err) }
|
||
if resp.Accepted != 1 { t.Fatal("expected accepted=1") }
|
||
if len(ms.events) != 1 { t.Fatal("event not submitted") }
|
||
}
|
||
|
||
func TestEventService_TrackEvent_InvalidEventType(t *testing.T) {
|
||
ms := &mockSink{}
|
||
svc := NewEventService(ms, []string{"asset.like"})
|
||
_, err := svc.TrackEvent(context.Background(), &model.Event{EventType: "evil.type"})
|
||
if err == nil { t.Fatal("expected error for invalid event type") }
|
||
}
|
||
|
||
func TestEventService_TrackEvent_PropertiesTooLarge(t *testing.T) {
|
||
ms := &mockSink{}
|
||
svc := NewEventService(ms, []string{"asset.like"})
|
||
big := make(map[string]string)
|
||
for i := 0; i < 2000; i++ { big[fmt.Sprintf("k%d", i)] = strings.Repeat("x", 100) }
|
||
_, err := svc.TrackEvent(context.Background(), &model.Event{
|
||
EventType: "asset.like", Properties: big,
|
||
})
|
||
if err == nil { t.Fatal("expected error for large properties") }
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 5.6: 跑测试（应失败）**
|
||
|
||
```bash
|
||
go test ./service/ -v
|
||
```
|
||
|
||
Expected: FAIL
|
||
|
||
- [ ] **Step 5.7: 实现 event_service**
|
||
|
||
`backend/services/statisticService/service/event_service.go`：
|
||
|
||
```go
|
||
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"time"
|
||
|
||
"github.com/topfans/backend/services/statisticService/model"
|
||
"github.com/topfans/backend/services/statisticService/sink"
|
||
pb "github.com/topfans/backend/pkg/proto/statistic"
|
||
)
|
||
|
||
var (
|
||
ErrInvalidEventType = fmt.Errorf("invalid event type")
|
||
ErrPropertiesTooLarge = fmt.Errorf("properties too large")
|
||
ErrInvalidEventID = fmt.Errorf("invalid event_id")
|
||
)
|
||
|
||
const MaxPropertiesSize = 1024 // 1KB
|
||
|
||
type EventService struct {
|
||
sink sink.EventSink
|
||
whiteList map[string]bool
|
||
statsEnable bool
|
||
}
|
||
|
||
func NewEventService(s sink.EventSink, whiteList []string) *EventService {
|
||
wl := make(map[string]bool, len(whiteList))
|
||
for _, t := range whiteList { wl[t] = true }
|
||
return &EventService{sink: s, whiteList: wl}
|
||
}
|
||
|
||
func (s *EventService) validate(e *model.Event) error {
|
||
if e.EventID == "" { return ErrInvalidEventID }
|
||
if !s.whiteList[e.EventType] { return ErrInvalidEventType }
|
||
b, _ := json.Marshal(e.Properties)
|
||
if len(b) > MaxPropertiesSize { return ErrPropertiesTooLarge }
|
||
return nil
|
||
}
|
||
|
||
func (s *EventService) TrackEvent(ctx context.Context, e *model.Event) (*pb.TrackEventResponse, error) {
|
||
if e.ReceivedAt.IsZero() { e.ReceivedAt = time.Now() }
|
||
if err := s.validate(e); err != nil { return nil, err }
|
||
if err := s.sink.Submit(ctx, e); err != nil {
|
||
return &pb.TrackEventResponse{Accepted: 0, Rejected: 1}, nil
|
||
}
|
||
return &pb.TrackEventResponse{Accepted: 1, Rejected: 0}, nil
|
||
}
|
||
|
||
func (s *EventService) BatchTrackEvent(ctx context.Context, es []*model.Event) (*pb.TrackEventResponse, error) {
|
||
accepted, rejected := 0, 0
|
||
for _, e := range es {
|
||
if e.ReceivedAt.IsZero() { e.ReceivedAt = time.Now() }
|
||
if err := s.validate(e); err != nil { rejected++; continue }
|
||
if err := s.sink.Submit(ctx, e); err != nil { rejected++ } else { accepted++ }
|
||
}
|
||
return &pb.TrackEventResponse{Accepted: int32(accepted), Rejected: int32(rejected)}, nil
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 5.8: 跑测试（应通过）**
|
||
|
||
```bash
|
||
go test ./service/ -v
|
||
```
|
||
|
||
Expected: PASS
|
||
|
||
- [ ] **Step 5.9: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/repository/ backend/services/statisticService/service/
|
||
git commit -m "feat(statistic): event_repo (batch insert + dedup) and event_service (validate + submit) (P2 T5)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 6: event_flusher Worker（攒批落库 + 触发 metric_recent_level_ups）
|
||
|
||
**Files:**
|
||
- Create: `backend/services/statisticService/worker/event_flusher.go`
|
||
- Create: `backend/services/statisticService/worker/event_flusher_test.go`
|
||
- Create: `backend/services/statisticService/repository/metric_repo.go`
|
||
|
||
- [ ] **Step 6.1: 写 metric_repo（同步更新 metric_recent_level_ups）**
|
||
|
||
`backend/services/statisticService/repository/metric_repo.go`：
|
||
|
||
```go
|
||
package repository
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"github.com/topfans/backend/services/statisticService/model"
|
||
)
|
||
|
||
type MetricRepository struct {
|
||
db *sql.DB
|
||
schema string
|
||
}
|
||
|
||
func NewMetricRepository(db *sql.DB, schema string) *MetricRepository {
|
||
return &MetricRepository{db: db, schema: schema}
|
||
}
|
||
|
||
func (r *MetricRepository) UpsertRecentLevelUp(ctx context.Context, e *model.Event) error {
|
||
if e.EventType != "asset.level_up" { return nil }
|
||
assetID := e.Properties["asset_id"]
|
||
fromLevel := e.Properties["from"]
|
||
toLevel := e.Properties["to"]
|
||
upgradeTime := e.OccurredAt
|
||
if assetID == "" || toLevel == "" { return nil }
|
||
_, err := r.db.ExecContext(ctx, fmt.Sprintf(`
|
||
INSERT INTO %s.metric_recent_level_ups
|
||
(user_id, star_id, asset_id, from_level, to_level, upgrade_time, asset_name, asset_thumb)
|
||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
||
ON CONFLICT DO NOTHING
|
||
`, r.schema),
|
||
e.UserID, e.StarID, assetID, fromLevel, toLevel, upgradeTime, "", "")
|
||
return err
|
||
}
|
||
```
|
||
|
||
> 完整 metric_repo 还含 weekly_user_income / upcoming_level_ups 读写，留到 T7/T8 任务。
|
||
|
||
- [ ] **Step 6.2: 写 event_flusher 测试**
|
||
|
||
`backend/services/statisticService/worker/event_flusher_test.go`：
|
||
|
||
```go
|
||
package worker
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"os"
|
||
"testing"
|
||
"time"
|
||
|
||
_ "github.com/lib/pq"
|
||
"github.com/topfans/backend/services/statisticService/model"
|
||
"github.com/topfans/backend/services/statisticService/repository"
|
||
)
|
||
|
||
func TestEventFlusher_FlushBatch(t *testing.T) {
|
||
dsn := os.Getenv("TEST_DATABASE_URL")
|
||
if dsn == "" { t.Skip("skip") }
|
||
db, _ := sql.Open("postgres", dsn)
|
||
defer db.Close()
|
||
|
||
eventRepo := repository.NewEventRepository(db, "statistic_test")
|
||
metricRepo := repository.NewMetricRepository(db, "statistic_test")
|
||
|
||
ch := make(chan *model.Event, 10)
|
||
flusher := NewEventFlusher(ch, eventRepo, metricRepo, 100, 1*time.Second)
|
||
|
||
go flusher.Start(context.Background())
|
||
|
||
for i := 0; i < 5; i++ {
|
||
ch <- &model.Event{EventID: fmt.Sprintf("u-%d", i), UserID: 1, StarID: 1, EventType: "asset.like",
|
||
OccurredAt: time.Now(), ReceivedAt: time.Now(), Properties: map[string]string{}}
|
||
}
|
||
time.Sleep(2 * time.Second) // 等攒批
|
||
flusher.Stop()
|
||
// 验证 events 表有 5 条
|
||
var n int
|
||
db.QueryRow("SELECT COUNT(*) FROM statistic_test.events").Scan(&n)
|
||
if n < 5 { t.Fatalf("expected >=5 events, got %d", n) }
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 6.3: 跑测试（应失败）**
|
||
|
||
```bash
|
||
TEST_DATABASE_URL="..." go test ./worker/ -v
|
||
```
|
||
|
||
Expected: FAIL
|
||
|
||
- [ ] **Step 6.4: 实现 event_flusher**
|
||
|
||
`backend/services/statisticService/worker/event_flusher.go`：
|
||
|
||
```go
|
||
package worker
|
||
|
||
import (
|
||
"context"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/topfans/backend/pkg/logger"
|
||
"github.com/topfans/backend/services/statisticService/metrics"
|
||
"github.com/topfans/backend/services/statisticService/model"
|
||
"github.com/topfans/backend/services/statisticService/repository"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
type EventFlusher struct {
|
||
ch <-chan *model.Event
|
||
eventRepo *repository.EventRepository
|
||
metricRepo *repository.MetricRepository
|
||
batchSize int
|
||
interval time.Duration
|
||
|
||
mu sync.Mutex
|
||
running bool
|
||
stop chan struct{}
|
||
}
|
||
|
||
func NewEventFlusher(ch <-chan *model.Event, eventRepo *repository.EventRepository, metricRepo *repository.MetricRepository, batchSize int, interval time.Duration) *EventFlusher {
|
||
return &EventFlusher{ch: ch, eventRepo: eventRepo, metricRepo: metricRepo, batchSize: batchSize, interval: interval, stop: make(chan struct{})}
|
||
}
|
||
|
||
func (f *EventFlusher) Start(ctx context.Context) {
|
||
f.mu.Lock()
|
||
f.running = true
|
||
f.mu.Unlock()
|
||
metrics.WorkerRunningCount.WithLabelValues("event_flusher").Set(1)
|
||
defer metrics.WorkerRunningCount.WithLabelValues("event_flusher").Set(0)
|
||
|
||
batch := make([]*model.Event, 0, f.batchSize)
|
||
ticker := time.NewTicker(f.interval)
|
||
defer ticker.Stop()
|
||
|
||
flush := func() {
|
||
if len(batch) == 0 { return }
|
||
inserted, err := f.eventRepo.InsertBatch(ctx, batch)
|
||
if err != nil {
|
||
logger.Logger.Error("event_flusher insert failed", zap.Error(err))
|
||
metrics.EventDBInsertTotal.WithLabelValues("failed").Inc()
|
||
} else {
|
||
metrics.EventDBInsertTotal.WithLabelValues("success").Inc()
|
||
}
|
||
// 同步触发 metric_recent_level_ups
|
||
for _, e := range batch {
|
||
if err := f.metricRepo.UpsertRecentLevelUp(ctx, e); err != nil {
|
||
logger.Logger.Warn("UpsertRecentLevelUp failed", zap.Error(err))
|
||
}
|
||
}
|
||
logger.Logger.Debug("event_flusher batch flushed", zap.Int("inserted", inserted), zap.Int("batch", len(batch)))
|
||
batch = batch[:0]
|
||
}
|
||
|
||
for {
|
||
select {
|
||
case <-f.stop:
|
||
flush()
|
||
return
|
||
case e := <-f.ch:
|
||
batch = append(batch, e)
|
||
metrics.EventChannelSize.Set(float64(len(f.ch)))
|
||
if len(batch) >= f.batchSize { flush() }
|
||
case <-ticker.C:
|
||
flush()
|
||
}
|
||
}
|
||
}
|
||
|
||
func (f *EventFlusher) Stop() {
|
||
f.mu.Lock()
|
||
defer f.mu.Unlock()
|
||
if f.running {
|
||
close(f.stop)
|
||
f.running = false
|
||
}
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 6.5: 跑测试（应通过）**
|
||
|
||
```bash
|
||
TEST_DATABASE_URL="..." go test ./worker/ -v
|
||
```
|
||
|
||
Expected: PASS
|
||
|
||
- [ ] **Step 6.6: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/repository/ backend/services/statisticService/worker/
|
||
git commit -m "feat(statistic): event_flusher worker (batch insert + sync metric_recent_level_ups) (P2 T6)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 7: metric_weekly_user_income_updater + metric_upcoming_level_ups_updater
|
||
|
||
**Files:**
|
||
- Modify: `backend/services/statisticService/repository/metric_repo.go`（加 weekly + upcoming 方法）
|
||
- Create: `backend/services/statisticService/worker/metric_weekly_user_income_updater.go`
|
||
- Create: `backend/services/statisticService/worker/metric_upcoming_level_ups_updater.go`
|
||
|
||
- [ ] **Step 7.1: 扩 metric_repo 加 weekly + upcoming 方法**
|
||
|
||
```go
|
||
// 加到 metric_repo.go
|
||
|
||
// RefreshWeeklyUserIncome: 全量重算本周 rank + total
|
||
func (r *MetricRepository) RefreshWeeklyUserIncome(ctx context.Context) error {
|
||
// 用 pg_try_advisory_lock 防多实例重复
|
||
var got bool
|
||
if err := r.db.QueryRowContext(ctx, "SELECT pg_try_advisory_lock(123456)").Scan(&got); err != nil {
|
||
return err
|
||
}
|
||
if !got { return nil } // 抢不到锁本轮跳过
|
||
defer r.db.ExecContext(ctx, "SELECT pg_advisory_unlock(123456)")
|
||
|
||
// 计算本周一 (Asia/Shanghai)
|
||
_, err := r.db.ExecContext(ctx, fmt.Sprintf(`
|
||
INSERT INTO %s.metric_weekly_user_income (star_id, user_id, week_start, total_crystal, rank_in_star)
|
||
SELECT
|
||
star_id, user_id,
|
||
DATE_TRUNC('week', received_at AT TIME ZONE 'Asia/Shanghai')::date AS week_start,
|
||
SUM(CASE WHEN event_type IN ('exhibition.revenue', 'crystal.change') AND (properties->>'amount')::BIGINT > 0
|
||
THEN (properties->>'amount')::BIGINT ELSE 0 END) AS total_crystal,
|
||
ROW_NUMBER() OVER (PARTITION BY star_id ORDER BY SUM(CASE WHEN event_type IN ('exhibition.revenue','crystal.change') AND (properties->>'amount')::BIGINT > 0 THEN (properties->>'amount')::BIGINT ELSE 0 END) DESC) AS rank_in_star
|
||
FROM %s.events
|
||
WHERE event_type IN ('exhibition.revenue', 'crystal.change')
|
||
AND received_at >= DATE_TRUNC('week', NOW() AT TIME ZONE 'Asia/Shanghai')
|
||
GROUP BY star_id, user_id
|
||
ON CONFLICT (star_id, user_id, week_start) DO UPDATE
|
||
SET total_crystal = EXCLUDED.total_crystal, rank_in_star = EXCLUDED.rank_in_star, updated_at = NOW()
|
||
`, r.schema, r.schema))
|
||
return err
|
||
}
|
||
|
||
// RefreshUpcomingLevelUps: 计算每个 asset 的 like_progress + duration_progress
|
||
func (r *MetricRepository) RefreshUpcomingLevelUps(ctx context.Context) error {
|
||
// 全量从 public.assets 读 + 计算进度
|
||
_, err := r.db.ExecContext(ctx, fmt.Sprintf(`
|
||
INSERT INTO %s.metric_upcoming_level_ups (user_id, star_id, asset_id, like_progress, duration_progress)
|
||
SELECT
|
||
a.user_id, a.star_id, a.id,
|
||
LEAST(100, (a.like_count::FLOAT / NULLIF(alc.upgrade_like_threshold, 0) * 100)::INT) AS like_progress,
|
||
LEAST(100, (EXTRACT(EPOCH FROM (NOW() - a.placed_at))::FLOAT / NULLIF(alc.upgrade_duration_seconds, 1) * 100)::INT) AS duration_progress
|
||
FROM public.assets a
|
||
JOIN public.asset_level_config alc ON alc.level = a.level
|
||
WHERE a.status = 'active' AND a.deleted_at IS NULL
|
||
ON CONFLICT (user_id, star_id, asset_id) DO UPDATE
|
||
SET like_progress = EXCLUDED.like_progress, duration_progress = EXCLUDED.duration_progress, updated_at = NOW()
|
||
`, r.schema))
|
||
return err
|
||
}
|
||
```
|
||
|
||
> **实现提示：** 实际项目里 `public.asset_level_config` 表名/字段名可能不同。P1 末向 assetService 同学确认后调整 SQL。
|
||
|
||
- [ ] **Step 7.2: 创建两个 Worker（结构同）**
|
||
|
||
`worker/metric_weekly_user_income_updater.go`：
|
||
|
||
```go
|
||
package worker
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"github.com/topfans/backend/pkg/logger"
|
||
"github.com/topfans/backend/services/statisticService/metrics"
|
||
"github.com/topfans/backend/services/statisticService/repository"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
type WeeklyUserIncomeUpdater struct {
|
||
repo *repository.MetricRepository
|
||
interval time.Duration
|
||
stop chan struct{}
|
||
}
|
||
|
||
func NewWeeklyUserIncomeUpdater(repo *repository.MetricRepository, interval time.Duration) *WeeklyUserIncomeUpdater {
|
||
return &WeeklyUserIncomeUpdater{repo: repo, interval: interval, stop: make(chan struct{})}
|
||
}
|
||
|
||
func (w *WeeklyUserIncomeUpdater) Start(ctx context.Context) {
|
||
metrics.WorkerRunningCount.WithLabelValues("weekly_user_income").Set(1)
|
||
defer metrics.WorkerRunningCount.WithLabelValues("weekly_user_income").Set(0)
|
||
ticker := time.NewTicker(w.interval)
|
||
defer ticker.Stop()
|
||
// 启动时跑一次
|
||
w.runOnce(ctx)
|
||
for {
|
||
select {
|
||
case <-w.stop:
|
||
return
|
||
case <-ticker.C:
|
||
w.runOnce(ctx)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (w *WeeklyUserIncomeUpdater) runOnce(ctx context.Context) {
|
||
t0 := time.Now()
|
||
if err := w.repo.RefreshWeeklyUserIncome(ctx); err != nil {
|
||
logger.Logger.Error("RefreshWeeklyUserIncome failed", zap.Error(err))
|
||
metrics.MVRefreshTotal.WithLabelValues("weekly_user_income", "failed").Inc()
|
||
} else {
|
||
metrics.MVRefreshTotal.WithLabelValues("weekly_user_income", "success").Inc()
|
||
}
|
||
metrics.MVRefreshDuration.WithLabelValues("weekly_user_income").Observe(time.Since(t0).Seconds())
|
||
}
|
||
|
||
func (w *WeeklyUserIncomeUpdater) Stop() { close(w.stop) }
|
||
```
|
||
|
||
`worker/metric_upcoming_level_ups_updater.go`：结构同上，调 `repo.RefreshUpcomingLevelUps`，interval 默认 15min。
|
||
|
||
- [ ] **Step 7.3: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/repository/ backend/services/statisticService/worker/
|
||
git commit -m "feat(statistic): metric_weekly + metric_upcoming level up workers (P2 T7)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 8: partitioner.go（events 分区自动管理）
|
||
|
||
**Files:**
|
||
- Create: `backend/services/statisticService/worker/partitioner.go`
|
||
- Create: `backend/services/statisticService/worker/partitioner_test.go`
|
||
|
||
- [ ] **Step 8.1: 写测试（集成）**
|
||
|
||
```go
|
||
// worker/partitioner_test.go
|
||
package worker
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"os"
|
||
"testing"
|
||
"time"
|
||
|
||
_ "github.com/lib/pq"
|
||
)
|
||
|
||
func TestPartitioner_EnsureFuture(t *testing.T) {
|
||
dsn := os.Getenv("TEST_DATABASE_URL")
|
||
if dsn == "" { t.Skip("skip") }
|
||
db, _ := sql.Open("postgres", dsn)
|
||
defer db.Close()
|
||
|
||
// 创建测试 schema
|
||
db.Exec("CREATE SCHEMA IF NOT EXISTS statistic_test")
|
||
db.Exec(`CREATE TABLE IF NOT EXISTS statistic_test.events (
|
||
id BIGSERIAL, event_id UUID NOT NULL, received_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||
PRIMARY KEY (id, received_at)
|
||
) PARTITION BY RANGE (received_at)`)
|
||
|
||
p := NewPartitioner(db, "statistic_test", 7, 30)
|
||
if err := p.EnsureFuturePartitions(context.Background(), 3); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// 验证未来 3 天分区存在
|
||
var n int
|
||
db.QueryRow("SELECT COUNT(*) FROM pg_tables WHERE schemaname='statistic_test' AND tablename LIKE 'events_%'").Scan(&n)
|
||
if n < 3 { t.Fatalf("expected >=3 partitions, got %d", n) }
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 8.2: 跑测试（应失败）**
|
||
|
||
```bash
|
||
TEST_DATABASE_URL="..." go test ./worker/ -v -run TestPartitioner
|
||
```
|
||
|
||
Expected: FAIL
|
||
|
||
- [ ] **Step 8.3: 实现 partitioner**
|
||
|
||
```go
|
||
// worker/partitioner.go
|
||
package worker
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"fmt"
|
||
"time"
|
||
|
||
"github.com/topfans/backend/pkg/logger"
|
||
"github.com/topfans/backend/services/statisticService/metrics"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
type Partitioner struct {
|
||
db *sql.DB
|
||
schema string
|
||
preCreateDays int
|
||
retentionDays int
|
||
stop chan struct{}
|
||
}
|
||
|
||
func NewPartitioner(db *sql.DB, schema string, preCreateDays, retentionDays int) *Partitioner {
|
||
return &Partitioner{db: db, schema: schema, preCreateDays: preCreateDays, retentionDays: retentionDays, stop: make(chan struct{})}
|
||
}
|
||
|
||
func (p *Partitioner) EnsureFuturePartitions(ctx context.Context, days int) error {
|
||
now := time.Now().In(time.FixedZone("Asia/Shanghai", 8*3600))
|
||
for i := 0; i <= days; i++ {
|
||
d := now.AddDate(0, 0, i)
|
||
next := d.AddDate(0, 0, 1)
|
||
name := fmt.Sprintf("events_%s", d.Format("2006_01_02"))
|
||
sql := fmt.Sprintf(`
|
||
CREATE TABLE IF NOT EXISTS %s.%s PARTITION OF %s.events
|
||
FOR VALUES FROM ('%s 00:00:00+08') TO ('%s 00:00:00+08')
|
||
`, p.schema, name, p.schema, d.Format("2006-01-02"), next.Format("2006-01-02"))
|
||
if _, err := p.db.ExecContext(ctx, sql); err != nil {
|
||
return fmt.Errorf("create partition %s: %w", name, err)
|
||
}
|
||
metrics.EventsPartitionCount.Inc()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (p *Partitioner) CleanupOldPartitions(ctx context.Context) error {
|
||
cutoff := time.Now().In(time.FixedZone("Asia/Shanghai", 8*3600)).AddDate(0, 0, -p.retentionDays)
|
||
rows, err := p.db.QueryContext(ctx, fmt.Sprintf(`
|
||
SELECT tablename FROM pg_tables
|
||
WHERE schemaname = $1 AND tablename LIKE 'events_%' AND tablename < $2
|
||
`, p.schema, fmt.Sprintf("events_%s", cutoff.Format("2006_01_02")))
|
||
if err != nil { return err }
|
||
defer rows.Close()
|
||
for rows.Next() {
|
||
var name string
|
||
rows.Scan(&name)
|
||
if _, err := p.db.ExecContext(ctx, fmt.Sprintf("DROP TABLE IF EXISTS %s.%s", p.schema, name)); err != nil {
|
||
logger.Logger.Warn("drop partition failed", zap.String("name", name), zap.Error(err))
|
||
} else {
|
||
logger.Logger.Info("dropped old partition", zap.String("name", name))
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (p *Partitioner) Start(ctx context.Context) {
|
||
// 启动时确保未来 7 天
|
||
p.EnsureFuturePartitions(ctx, p.preCreateDays)
|
||
ticker := time.NewTicker(1 * time.Hour)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-p.stop:
|
||
return
|
||
case <-ticker.C:
|
||
hour := time.Now().Hour()
|
||
if hour == 0 && time.Now().Minute() < 10 {
|
||
p.EnsureFuturePartitions(ctx, p.preCreateDays)
|
||
}
|
||
if hour == 0 && time.Now().Minute() >= 30 && time.Now().Minute() < 40 {
|
||
p.CleanupOldPartitions(ctx)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (p *Partitioner) Stop() { close(p.stop) }
|
||
```
|
||
|
||
- [ ] **Step 8.4: 跑测试（应通过）**
|
||
|
||
```bash
|
||
TEST_DATABASE_URL="..." go test ./worker/ -v -run TestPartitioner
|
||
```
|
||
|
||
Expected: PASS
|
||
|
||
- [ ] **Step 8.5: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/worker/partitioner.go backend/services/statisticService/worker/partitioner_test.go
|
||
git commit -m "feat(statistic): partitioner worker (auto create/drop events partitions) (P2 T8)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 9: Provider (TrackEvent/BatchTrackEvent) + main.go 集成 + socialService 联调
|
||
|
||
**Files:**
|
||
- Create: `backend/services/statisticService/provider/statistic_internal_provider.go`
|
||
- Modify: `backend/services/statisticService/main.go`
|
||
- Create: `pkg/statistic/client.go`（业务侧统一 SDK）
|
||
- Modify: `backend/services/socialService/service/asset_like_service.go`（联调）
|
||
|
||
- [ ] **Step 9.1: 写 provider**
|
||
|
||
`backend/services/statisticService/provider/statistic_internal_provider.go`：
|
||
|
||
```go
|
||
package provider
|
||
|
||
import (
|
||
"context"
|
||
|
||
"github.com/topfans/backend/pkg/logger"
|
||
pb "github.com/topfans/backend/pkg/proto/statistic"
|
||
"github.com/topfans/backend/pkg/proto/event"
|
||
"github.com/topfans/backend/services/statisticService/model"
|
||
"github.com/topfans/backend/services/statisticService/service"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
type StatisticInternalProvider struct {
|
||
eventSvc *service.EventService
|
||
}
|
||
|
||
func NewStatisticInternalProvider(eventSvc *service.EventService) *StatisticInternalProvider {
|
||
return &StatisticInternalProvider{eventSvc: eventSvc}
|
||
}
|
||
|
||
func (p *StatisticInternalProvider) TrackEvent(ctx context.Context, e *event.Event) (*pb.TrackEventResponse, error) {
|
||
return p.eventSvc.TrackEvent(ctx, toModel(e))
|
||
}
|
||
|
||
func (p *StatisticInternalProvider) BatchTrackEvent(ctx context.Context, req *event.BatchEventRequest) (*pb.TrackEventResponse, error) {
|
||
events := make([]*model.Event, 0, len(req.Events))
|
||
for _, e := range req.Events { events = append(events, toModel(e)) }
|
||
return p.eventSvc.BatchTrackEvent(ctx, events)
|
||
}
|
||
|
||
func toModel(e *event.Event) *model.Event {
|
||
return &model.Event{
|
||
EventID: e.EventId, UserID: e.UserId, StarID: e.StarId, EventType: e.EventType,
|
||
OccurredAt: time.UnixMilli(e.OccurredAt), ReceivedAt: time.UnixMilli(e.ReceivedAt),
|
||
Properties: e.Properties,
|
||
}
|
||
}
|
||
```
|
||
|
||
补 `import "time"`。
|
||
|
||
- [ ] **Step 9.2: 写 main.go 集成（P2 完整版）**
|
||
|
||
替换 P1 骨架 main.go，加上：
|
||
- 创建 ChannelEventSink
|
||
- 创建 EventRepository + MetricRepository
|
||
- 创建 EventService
|
||
- 创建 EventFlusher + Partitioner + WeeklyUserIncomeUpdater + UpcomingLevelUpsUpdater
|
||
- 启动 4 个 worker（goroutine）
|
||
- 创建 StatisticInternalProvider
|
||
- Dubbo 注册 + Serve
|
||
|
||
```go
|
||
// 加到 main.go (替换 Step 1.5 占位 main.go)
|
||
|
||
// ... 前面 init logger/config/db/redis/healthz 不变 ...
|
||
|
||
eventCh := make(chan *model.Event, config.ChannelCfg.EventChannelCapacity)
|
||
metrics.EventChannelCapacity.Set(float64(config.ChannelCfg.EventChannelCapacity))
|
||
|
||
// 启动 partitioner
|
||
partitioner := worker.NewPartitioner(db, config.DBConfig.Schema, config.PartitionCfg.PreCreateDays, config.PartitionCfg.RetentionDays)
|
||
go partitioner.Start(context.Background())
|
||
|
||
// repository
|
||
eventRepo := repository.NewEventRepository(db, config.DBConfig.Schema)
|
||
metricRepo := repository.NewMetricRepository(db, config.DBConfig.Schema)
|
||
|
||
// sink
|
||
cs := sink.NewChannelEventSink(eventCh)
|
||
|
||
// service
|
||
whiteList := []string{"asset.like", "asset.mint", "exhibition.start", "exhibition.end",
|
||
"exhibition.revenue", "asset.level_up", "crystal.change"}
|
||
eventSvc := service.NewEventService(cs, whiteList)
|
||
|
||
// workers
|
||
flusher := worker.NewEventFlusher(eventCh, eventRepo, metricRepo, config.ChannelCfg.EventBatchSize, config.ChannelCfg.EventBatchInterval)
|
||
go flusher.Start(context.Background())
|
||
|
||
weeklyW := worker.NewWeeklyUserIncomeUpdater(metricRepo, config.RefreshCfg.WeeklyUserIncome)
|
||
go weeklyW.Start(context.Background())
|
||
|
||
upcomingW := worker.NewUpcomingLevelUpsUpdater(metricRepo, config.RefreshCfg.UpcomingLevelUps)
|
||
go upcomingW.Start(context.Background())
|
||
|
||
// provider
|
||
internalProvider := provider.NewStatisticInternalProvider(eventSvc)
|
||
|
||
// Dubbo server
|
||
srv, _ := server.NewServer(server.WithServerProtocol(protocol.WithPort(*port), protocol.WithTriple()))
|
||
pb.RegisterStatisticServiceHandler(srv, internalProvider)
|
||
go srv.Serve()
|
||
|
||
// ... graceful shutdown ...
|
||
```
|
||
|
||
- [ ] **Step 9.3: 编译验证**
|
||
|
||
```bash
|
||
go build ./services/statisticService
|
||
```
|
||
|
||
Expected: 编译成功
|
||
|
||
- [ ] **Step 9.4: 创建 pkg/statistic 业务侧 SDK**
|
||
|
||
`backend/pkg/statistic/client.go`：
|
||
|
||
```go
|
||
package statistic
|
||
|
||
import (
|
||
"context"
|
||
"sync"
|
||
"time"
|
||
|
||
"dubbo.apache.org/dubbo-go/v3/client"
|
||
"github.com/google/uuid"
|
||
pb "github.com/topfans/backend/pkg/proto/event"
|
||
)
|
||
|
||
var (
|
||
instance *Client
|
||
once sync.Once
|
||
)
|
||
|
||
type Client struct {
|
||
service pb.StatisticService
|
||
}
|
||
|
||
func Init(dubboClient *client.Client) error {
|
||
var err error
|
||
once.Do(func() {
|
||
svc, e := pb.NewStatisticService(dubboClient)
|
||
if e != nil { err = e; return }
|
||
instance = &Client{service: svc}
|
||
})
|
||
return err
|
||
}
|
||
|
||
func Get() *Client { return instance }
|
||
|
||
// TrackEvent fire-and-forget, 不阻塞业务
|
||
func (c *Client) TrackEvent(ctx context.Context, e *pb.Event) {
|
||
if c == nil || c.service == nil { return }
|
||
if e.EventId == "" { e.EventId = uuid.New().String() }
|
||
if e.OccurredAt == 0 { e.OccurredAt = time.Now().UnixMilli() }
|
||
// 用独立 context 避免业务 ctx cancel 影响
|
||
bgCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
go func() {
|
||
defer cancel()
|
||
c.service.TrackEvent(bgCtx, e)
|
||
}()
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 9.5: 业务侧 socialService 集成**
|
||
|
||
修改 `backend/services/socialService/service/asset_like_service.go`，在 `LikeAsset` 写 `like_income_log` 后加：
|
||
|
||
```go
|
||
// 写 like_income_log 之后
|
||
statistic.Get().TrackEvent(context.Background(), &eventPb.Event{
|
||
EventId: uuid.New().String(),
|
||
UserId: userID,
|
||
StarId: starID,
|
||
EventType: "asset.like",
|
||
OccurredAt: time.Now().UnixMilli(),
|
||
Properties: map[string]string{
|
||
"asset_id": strconv.FormatInt(assetID, 10),
|
||
"amount": strconv.FormatInt(amount, 10),
|
||
},
|
||
})
|
||
```
|
||
|
||
并在 socialService main.go 调用 `statistic.Init(statisticClient)`。
|
||
|
||
- [ ] **Step 9.6: 启动两端 + 联调验证**
|
||
|
||
```bash
|
||
# Terminal 1
|
||
go run ./services/statisticService
|
||
|
||
# Terminal 2
|
||
go run ./services/socialService
|
||
|
||
# Terminal 3: 触发点赞
|
||
# 用项目现有的 e2e 脚本或 API 调用
|
||
```
|
||
|
||
Expected:
|
||
- `statistic.events` 表有新行（`event_type='asset.like'`）
|
||
- `metric_recent_level_ups` 表无变化（因为是 asset.like 不是 asset.level_up）
|
||
- 日志显示 `event_flusher batch flushed`
|
||
|
||
- [ ] **Step 9.7: 跑 P2 全量测试**
|
||
|
||
```bash
|
||
cd backend
|
||
TEST_DATABASE_URL="..." go test ./services/statisticService/... -v
|
||
```
|
||
|
||
Expected: 全部通过
|
||
|
||
- [ ] **Step 9.8: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/ backend/pkg/statistic/ backend/services/socialService/
|
||
git commit -m "feat(statistic): P2 complete - event collection framework + socialService integration (T9)"
|
||
```
|
||
|
||
---
|
||
|
||
## P3 看板 7 RPC
|
||
|
||
### Task 10: 4 个物化视图 Worker（materializer）
|
||
|
||
**Files:**
|
||
- Create: `backend/services/statisticService/worker/materializer.go`
|
||
- Create: `backend/services/statisticService/worker/materializer_test.go`
|
||
|
||
> 4 个 MV 已由 T3 步骤 3.5 创建 DDL。本任务实现定时刷新逻辑。
|
||
|
||
- [ ] **Step 10.1: 写 materializer 测试**
|
||
|
||
```go
|
||
// worker/materializer_test.go
|
||
package worker
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"os"
|
||
"testing"
|
||
"time"
|
||
|
||
_ "github.com/lib/pq"
|
||
)
|
||
|
||
func TestMaterializer_RefreshDailyUserIncome(t *testing.T) {
|
||
dsn := os.Getenv("TEST_DATABASE_URL")
|
||
if dsn == "" { t.Skip("skip") }
|
||
db, _ := sql.Open("postgres", dsn)
|
||
defer db.Close()
|
||
|
||
m := NewMaterializer(db, "statistic_test", 5*time.Second)
|
||
if err := m.RefreshOne(context.Background(), "mv_daily_user_income"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// 验证 refresh_log 有记录
|
||
var n int
|
||
db.QueryRow("SELECT COUNT(*) FROM statistic_test.refresh_log WHERE mv_name='mv_daily_user_income'").Scan(&n)
|
||
if n < 1 { t.Fatal("no refresh_log entry") }
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 10.2: 实现 materializer**
|
||
|
||
```go
|
||
// worker/materializer.go
|
||
package worker
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"time"
|
||
|
||
"github.com/topfans/backend/pkg/logger"
|
||
"github.com/topfans/backend/services/statisticService/metrics"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
type Materializer struct {
|
||
db *sql.DB
|
||
schema string
|
||
stop chan struct{}
|
||
}
|
||
|
||
func NewMaterializer(db *sql.DB, schema string, interval time.Duration) *Materializer {
|
||
return &Materializer{db: db, schema: schema, stop: make(chan struct{})}
|
||
}
|
||
|
||
var mvList = []string{
|
||
"mv_daily_user_income",
|
||
"mv_daily_exhibition_revenue",
|
||
"mv_daily_like_income",
|
||
"mv_asset_level_distribution",
|
||
}
|
||
|
||
func (m *Materializer) RefreshOne(ctx context.Context, mvName string) error {
|
||
var got bool
|
||
if err := m.db.QueryRowContext(ctx, "SELECT pg_try_advisory_lock(234567)").Scan(&got); err != nil {
|
||
return err
|
||
}
|
||
if !got { return nil }
|
||
defer m.db.ExecContext(ctx, "SELECT pg_advisory_unlock(234567)")
|
||
|
||
t0 := time.Now()
|
||
var mvID int
|
||
if err := m.db.QueryRowContext(ctx,
|
||
`INSERT INTO `+m.schema+`.refresh_log (mv_name, started_at, status) VALUES ($1, NOW(), 'running') RETURNING id`,
|
||
mvName).Scan(&mvID); err != nil {
|
||
return err
|
||
}
|
||
|
||
_, err := m.db.ExecContext(ctx, "REFRESH MATERIALIZED VIEW CONCURRENTLY "+m.schema+"."+mvName)
|
||
if err != nil {
|
||
m.db.ExecContext(ctx, `UPDATE `+m.schema+`.refresh_log SET status='failed', finished_at=NOW(), error_message=$1 WHERE id=$2`, err.Error(), mvID)
|
||
metrics.MVRefreshTotal.WithLabelValues(mvName, "failed").Inc()
|
||
return err
|
||
}
|
||
m.db.ExecContext(ctx, `UPDATE `+m.schema+`.refresh_log SET status='success', finished_at=NOW() WHERE id=$1`, mvID)
|
||
metrics.MVRefreshTotal.WithLabelValues(mvName, "success").Inc()
|
||
metrics.MVRefreshDuration.WithLabelValues(mvName).Observe(time.Since(t0).Seconds())
|
||
return nil
|
||
}
|
||
|
||
func (m *Materializer) Start(ctx context.Context, interval time.Duration) {
|
||
metrics.WorkerRunningCount.WithLabelValues("materializer").Set(1)
|
||
defer metrics.WorkerRunningCount.WithLabelValues("materializer").Set(0)
|
||
// 启动时错开 30s 跑一次每个
|
||
for i, mv := range mvList {
|
||
go func(idx int, mvName string) {
|
||
time.Sleep(time.Duration(idx*30) * time.Second)
|
||
for {
|
||
select {
|
||
case <-m.stop: return
|
||
case <-time.After(interval):
|
||
if err := m.RefreshOne(ctx, mvName); err != nil {
|
||
logger.Logger.Error("RefreshOne failed", zap.String("mv", mvName), zap.Error(err))
|
||
}
|
||
}
|
||
}
|
||
}(i, mv)
|
||
}
|
||
}
|
||
|
||
func (m *Materializer) Stop() { close(m.stop) }
|
||
```
|
||
|
||
- [ ] **Step 10.3: 跑测试（应通过）**
|
||
|
||
```bash
|
||
TEST_DATABASE_URL="..." go test ./worker/ -v -run TestMaterializer
|
||
```
|
||
|
||
Expected: PASS
|
||
|
||
- [ ] **Step 10.4: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/worker/materializer.go
|
||
git commit -m "feat(statistic): materializer worker (4 MV refresh with concurrent + advisory lock) (P3 T10)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 11: dashboard_repo（7 聚合 SQL）+ dashboard_service（业务逻辑 + 缓存）
|
||
|
||
**Files:**
|
||
- Create: `backend/services/statisticService/repository/dashboard_repo.go`
|
||
- Create: `backend/services/statisticService/service/dashboard_service.go`
|
||
- Create: `backend/services/statisticService/service/cache.go`
|
||
|
||
- [ ] **Step 11.1: 写 dashboard_repo 7 个 SQL 方法 + 测试**
|
||
|
||
每个方法一个 SQL，对应 spec §2.3 + §3.4/3.5 视图。示例：
|
||
|
||
```go
|
||
// repository/dashboard_repo.go
|
||
package repository
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"time"
|
||
)
|
||
|
||
type DashboardRepository struct {
|
||
db *sql.DB
|
||
schema string
|
||
}
|
||
|
||
func NewDashboardRepository(db *sql.DB, schema string) *DashboardRepository {
|
||
return &DashboardRepository{db: db, schema: schema}
|
||
}
|
||
|
||
// GetTodayOverviewPart: 查 metric_weekly_user_income 算 week_rank
|
||
type TodayOverviewPart struct {
|
||
WeekRank int
|
||
WeekTotalUsers int
|
||
}
|
||
|
||
func (r *DashboardRepository) GetWeekRank(ctx context.Context, userID, starID int64) (*TodayOverviewPart, error) {
|
||
weekStart := weekStartMonday(time.Now())
|
||
var rank sql.NullInt64
|
||
err := r.db.QueryRowContext(ctx, fmt.Sprintf(`
|
||
SELECT rank_in_star FROM %s.metric_weekly_user_income
|
||
WHERE star_id=$1 AND user_id=$2 AND week_start=$3
|
||
`, r.schema), starID, userID, weekStart).Scan(&rank)
|
||
if err == sql.ErrNoRows { return &TodayOverviewPart{WeekRank: -1, WeekTotalUsers: 0}, nil }
|
||
if err != nil { return nil, err }
|
||
|
||
var totalUsers int
|
||
err = r.db.QueryRowContext(ctx, fmt.Sprintf(`
|
||
SELECT COUNT(*) FROM %s.metric_weekly_user_income
|
||
WHERE star_id=$1 AND week_start=$2 AND total_crystal > 0
|
||
`, r.schema), starID, weekStart).Scan(&totalUsers)
|
||
if err != nil { return nil, err }
|
||
|
||
return &TodayOverviewPart{
|
||
WeekRank: int(rank.Int64),
|
||
WeekTotalUsers: totalUsers,
|
||
}, nil
|
||
}
|
||
|
||
func (r *DashboardRepository) GetTodayIncome(ctx context.Context, userID, starID int64) (int64, error) {
|
||
var income sql.NullInt64
|
||
err := r.db.QueryRowContext(ctx, fmt.Sprintf(`
|
||
SELECT COALESCE(SUM(CASE WHEN (properties->>'amount')::BIGINT > 0
|
||
THEN (properties->>'amount')::BIGINT ELSE 0 END), 0)
|
||
FROM %s.events
|
||
WHERE user_id=$1 AND star_id=$2
|
||
AND event_type IN ('exhibition.revenue', 'crystal.change')
|
||
AND received_at >= DATE_TRUNC('day', NOW() AT TIME ZONE 'Asia/Shanghai')
|
||
`, r.schema), userID, starID).Scan(&income)
|
||
if err != nil && err != sql.ErrNoRows { return 0, err }
|
||
return income.Int64, nil
|
||
}
|
||
|
||
// 7 日收益曲线: 读 mv_daily_user_income
|
||
type DailyIncomePoint struct {
|
||
Date string
|
||
Income int64
|
||
}
|
||
|
||
func (r *DashboardRepository) Get7DayIncomeCurve(ctx context.Context, userID, starID int64) ([]DailyIncomePoint, int64, error) {
|
||
rows, err := r.db.QueryContext(ctx, fmt.Sprintf(`
|
||
SELECT income_date::text, COALESCE(total_crystal, 0) FROM %s.mv_daily_user_income
|
||
WHERE user_id=$1 AND star_id=$2
|
||
AND income_date >= (DATE_TRUNC('day', NOW() AT TIME ZONE 'Asia/Shanghai') - INTERVAL '6 days')::date
|
||
ORDER BY income_date ASC
|
||
`, r.schema), userID, starID)
|
||
if err != nil { return nil, 0, err }
|
||
defer rows.Close()
|
||
var points []DailyIncomePoint
|
||
var total int64
|
||
for rows.Next() {
|
||
var p DailyIncomePoint
|
||
rows.Scan(&p.Date, &p.Income)
|
||
points = append(points, p)
|
||
total += p.Income
|
||
}
|
||
return points, total, nil
|
||
}
|
||
|
||
// 剩余 5 个 RPC 类似实现...
|
||
|
||
func weekStartMonday(t time.Time) time.Time {
|
||
loc, _ := time.LoadLocation("Asia/Shanghai")
|
||
t2 := t.In(loc)
|
||
offset := (int(t2.Weekday()) + 6) % 7 // Mon=0
|
||
return time.Date(t2.Year(), t2.Month(), t2.Day()-offset, 0, 0, 0, 0, loc)
|
||
}
|
||
```
|
||
|
||
测试：每个方法一个集成测试，用 `TEST_DATABASE_URL` skip 模式，构造测试数据 + 验证返回值。
|
||
|
||
- [ ] **Step 11.2: 写 cache 封装**
|
||
|
||
```go
|
||
// service/cache.go
|
||
package service
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
)
|
||
|
||
type Cache struct {
|
||
rdb *redis.Client
|
||
ttl time.Duration
|
||
}
|
||
|
||
func NewCache(rdb *redis.Client) *Cache { return &Cache{rdb: rdb, ttl: 5 * time.Minute} }
|
||
|
||
func (c *Cache) Get(ctx context.Context, key string) (string, bool, error) {
|
||
v, err := c.rdb.Get(ctx, key).Result()
|
||
if err == redis.Nil { return "", false, nil }
|
||
if err != nil { return "", false, err }
|
||
return v, true, nil
|
||
}
|
||
|
||
func (c *Cache) Set(ctx context.Context, key, value string) error {
|
||
return c.rdb.Set(ctx, key, value, c.ttl).Err()
|
||
}
|
||
|
||
func (c *Cache) SetEmpty(ctx context.Context, key string) error {
|
||
// 缓存穿透防护：空值 1 分钟
|
||
return c.rdb.Set(ctx, key, "null", 1*time.Minute).Err()
|
||
}
|
||
|
||
func CacheKey(rpc string, starID, userID int64) string {
|
||
return fmt.Sprintf("dash:%s:%d:%d", rpc, starID, userID)
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 11.3: 写 dashboard_service（7 RPC 业务逻辑 + 缓存 + 降级）**
|
||
|
||
```go
|
||
// service/dashboard_service.go
|
||
package service
|
||
|
||
import (
|
||
"context"
|
||
|
||
"github.com/topfans/backend/pkg/logger"
|
||
pb "github.com/topfans/backend/pkg/proto/statistic"
|
||
"github.com/topfans/backend/services/statisticService/repository"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
type DashboardService struct {
|
||
repo *repository.DashboardRepository
|
||
cache *Cache
|
||
userRPC UserRPCClient // 跨服务调 userService.crystal_balance
|
||
}
|
||
|
||
func NewDashboardService(repo *repository.DashboardRepository, cache *Cache, userRPC UserRPCClient) *DashboardService {
|
||
return &DashboardService{repo: repo, cache: cache, userRPC: userRPC}
|
||
}
|
||
|
||
func (s *DashboardService) GetTodayOverview(ctx context.Context, userID, starID int64) (*pb.GetTodayOverviewResponse, error) {
|
||
key := CacheKey("today_overview", starID, userID)
|
||
if v, ok, _ := s.cache.Get(ctx, key); ok { /* unmarshal & return */ }
|
||
|
||
part, err := s.repo.GetWeekRank(ctx, userID, starID)
|
||
if err != nil { return nil, err }
|
||
todayIncome, err := s.repo.GetTodayIncome(ctx, userID, starID)
|
||
if err != nil { return nil, err }
|
||
|
||
// 跨服务: userService.crystal_balance
|
||
var crystal int64
|
||
if s.userRPC != nil {
|
||
if c, err := s.userRPC.GetCrystalBalance(ctx, userID, starID); err == nil {
|
||
crystal = c
|
||
} else {
|
||
logger.Logger.Warn("userService.GetCrystalBalance failed, use 0", zap.Error(err))
|
||
// 降级：返回 0，stale 标记
|
||
}
|
||
}
|
||
|
||
resp := &pb.GetTodayOverviewResponse{
|
||
CrystalBalance: crystal,
|
||
TodayIncome: todayIncome,
|
||
WeekRank: int32(part.WeekRank),
|
||
WeekTotalUsers: int32(part.WeekTotalUsers),
|
||
}
|
||
// 缓存 5min (省略 marshal)
|
||
return resp, nil
|
||
}
|
||
|
||
// 剩余 6 个 RPC 类似...
|
||
|
||
// UserRPCClient interface
|
||
type UserRPCClient interface {
|
||
GetCrystalBalance(ctx context.Context, userID, starID int64) (int64, error)
|
||
}
|
||
```
|
||
|
||
> 完整 6 个 RPC 实现遵循同样模式：先查 cache → miss 时查 MV/预聚表 → 构造响应 → 缓存。代码量大但模式统一。
|
||
|
||
- [ ] **Step 11.4: 跑 dashboard_service 测试**
|
||
|
||
```bash
|
||
TEST_DATABASE_URL="..." go test ./service/ -v -run TestDashboard
|
||
```
|
||
|
||
Expected: 7 个 RPC 单测全部通过
|
||
|
||
- [ ] **Step 11.5: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/repository/dashboard_repo.go \
|
||
backend/services/statisticService/service/dashboard_service.go \
|
||
backend/services/statisticService/service/cache.go
|
||
git commit -m "feat(statistic): dashboard_repo (7 SQL) + dashboard_service (7 RPC + cache + degrade) (P3 T11)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 12: 看板 mobile provider + main.go 集成
|
||
|
||
**Files:**
|
||
- Create: `backend/services/statisticService/provider/statistic_mobile_provider.go`
|
||
- Modify: `backend/services/statisticService/main.go`
|
||
|
||
- [ ] **Step 12.1: 写 mobile provider**
|
||
|
||
```go
|
||
// provider/statistic_mobile_provider.go
|
||
package provider
|
||
|
||
import (
|
||
"context"
|
||
|
||
"github.com/topfans/backend/pkg/logger"
|
||
"github.com/topfans/backend/services/statisticService/metrics"
|
||
pb "github.com/topfans/backend/pkg/proto/statistic"
|
||
"github.com/topfans/backend/services/statisticService/service"
|
||
"go.uber.org/zap"
|
||
"strconv"
|
||
"time"
|
||
)
|
||
|
||
type StatisticMobileProvider struct {
|
||
dashSvc *service.DashboardService
|
||
}
|
||
|
||
func NewStatisticMobileProvider(dashSvc *service.DashboardService) *StatisticMobileProvider {
|
||
return &StatisticMobileProvider{dashSvc: dashSvc}
|
||
}
|
||
|
||
func userIDFromContext(ctx context.Context) int64 {
|
||
if v := ctx.Value("user_id"); v != nil {
|
||
if s, ok := v.(string); ok { n, _ := strconv.ParseInt(s, 10, 64); return n }
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func (p *StatisticMobileProvider) recordRPC(rpc string, start time.Time, err error) {
|
||
status := "ok"
|
||
if err != nil { status = "error" }
|
||
metrics.DashboardRPCTotal.WithLabelValues(rpc, status).Inc()
|
||
metrics.DashboardRPCDuration.WithLabelValues(rpc).Observe(time.Since(start).Seconds())
|
||
}
|
||
|
||
func (p *StatisticMobileProvider) GetTodayOverview(ctx context.Context, req *pb.GetTodayOverviewRequest) (*pb.GetTodayOverviewResponse, error) {
|
||
t0 := time.Now(); defer func() { p.recordRPC("GetTodayOverview", t0, nil) }()
|
||
return p.dashSvc.GetTodayOverview(ctx, userIDFromContext(ctx), req.StarId)
|
||
}
|
||
|
||
// 剩余 6 个 RPC 方法类似...
|
||
```
|
||
|
||
- [ ] **Step 12.2: 修改 main.go 注册 mobile provider**
|
||
|
||
加：
|
||
```go
|
||
// 加到 main.go (在 internalProvider 后)
|
||
|
||
dashRepo := repository.NewDashboardRepository(db, config.DBConfig.Schema)
|
||
cache := service.NewCache(rdb)
|
||
userCli, _ := dubboclient.NewClient(dubboclient.WithClientURL("tri://localhost:20000")) // userService
|
||
userRPC := client.NewUserServiceClient(pbUser.NewUserSocialService(userCli))
|
||
dashSvc := service.NewDashboardService(dashRepo, cache, userRPC)
|
||
|
||
mobileProvider := provider.NewStatisticMobileProvider(dashSvc)
|
||
pb.RegisterStatisticServiceHandler(srv, mobileProvider)
|
||
```
|
||
|
||
> **注意**：原 proto 是 `service StatisticService`，包含 7 看板 + 2 事件 = 9 个 RPC。需在 statistic.proto 里把 mobile 和 internal 拆成两个 service（`StatisticMobileService` + `StatisticInternalService`），否则会冲突。本步骤前提：T2 已拆分。**如果没拆，需要回到 T2 调整 proto，重新生成。**
|
||
|
||
实际项目更可能是两个 service，本计划假设已拆（spec §1.3 提到 mobile_provider / internal_provider 两个文件）。
|
||
|
||
- [ ] **Step 12.3: 编译 + 启动 + 测 7 RPC**
|
||
|
||
```bash
|
||
go build ./services/statisticService
|
||
go run ./services/statisticService &
|
||
# 7 RPC 端到端测试
|
||
# 用 grpcurl 或写个简单的 Go client
|
||
```
|
||
|
||
Expected: 7 个 RPC 都返回非错误响应
|
||
|
||
- [ ] **Step 12.4: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/
|
||
git commit -m "feat(statistic): mobile provider (7 RPC) + main.go 集成 (P3 T12)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 13: Gateway 7 路由 + controller
|
||
|
||
**Files:**
|
||
- Modify: `backend/gateway/router/router.go`
|
||
- Create: `backend/gateway/controller/statistic_controller.go`
|
||
|
||
- [ ] **Step 13.1: 创建 controller**
|
||
|
||
```go
|
||
// gateway/controller/statistic_controller.go
|
||
package controller
|
||
|
||
import (
|
||
"context"
|
||
"net/http"
|
||
"strconv"
|
||
"time"
|
||
|
||
"dubbo.apache.org/dubbo-go/v3/client"
|
||
"github.com/gin-gonic/gin"
|
||
"github.com/topfans/backend/gateway/pkg/response"
|
||
pb "github.com/topfans/backend/pkg/proto/statistic"
|
||
)
|
||
|
||
type StatisticController struct {
|
||
mobile pb.StatisticMobileService
|
||
}
|
||
|
||
func NewStatisticController(dubboClient *client.Client) (*StatisticController, error) {
|
||
svc, err := pb.NewStatisticMobileService(dubboClient)
|
||
if err != nil { return nil, err }
|
||
return &StatisticController{mobile: svc}, nil
|
||
}
|
||
|
||
func (ctrl *StatisticController) parseStarID(c *gin.Context) (int64, bool) {
|
||
v := c.Query("star_id")
|
||
if v == "" { response.Error(c, 400, "star_id is required"); return 0, false }
|
||
n, err := strconv.ParseInt(v, 10, 64)
|
||
if err != nil { response.Error(c, 400, "star_id invalid"); return 0, false }
|
||
return n, true
|
||
}
|
||
|
||
func (ctrl *StatisticController) ctxWithUserID(c *gin.Context) context.Context {
|
||
ctx, cancel := context.WithTimeout(c.Request.Context(), 5*time.Second)
|
||
defer cancel()
|
||
userID, _ := c.Get("user_id")
|
||
// 透传到 dubbo 调 user_id
|
||
return ctxWithUserIDValue(ctx, userID)
|
||
}
|
||
|
||
// 7 个 controller 方法
|
||
func (ctrl *StatisticController) GetTodayOverview(c *gin.Context) {
|
||
starID, ok := ctrl.parseStarID(c); if !ok { return }
|
||
resp, err := ctrl.mobile.GetTodayOverview(ctrl.ctxWithUserID(c), &pb.GetTodayOverviewRequest{StarId: starID})
|
||
if err != nil { response.Error(c, 500, err.Error()); return }
|
||
response.OK(c, gin.H{"code": 200, "data": resp})
|
||
}
|
||
|
||
// 6 个类似...
|
||
```
|
||
|
||
- [ ] **Step 13.2: 注册路由**
|
||
|
||
修改 `backend/gateway/router/router.go`，加：
|
||
|
||
```go
|
||
// 在 SetupRouter 函数里
|
||
statisticCtrl, err := controller.NewStatisticController(statisticClient)
|
||
if err != nil { return nil, err }
|
||
api := r.Group("/api/v1/dashboard")
|
||
{
|
||
api.GET("/today-overview", middleware.JWTAuth(), statisticCtrl.GetTodayOverview)
|
||
api.GET("/income-curve", middleware.JWTAuth(), statisticCtrl.Get7DayIncomeCurve)
|
||
api.GET("/exhibition-summary", middleware.JWTAuth(), statisticCtrl.GetExhibitionIncomeSummary)
|
||
api.GET("/like-income-by-level", middleware.JWTAuth(), statisticCtrl.GetLikeIncomeByLevel)
|
||
api.GET("/top-assets", middleware.JWTAuth(), statisticCtrl.GetTopAssetsByEarning)
|
||
api.GET("/level-distribution", middleware.JWTAuth(), statisticCtrl.GetAssetLevelDistribution)
|
||
api.GET("/upgrade-progress", middleware.JWTAuth(), statisticCtrl.GetAssetUpgradeProgress)
|
||
}
|
||
```
|
||
|
||
并在 SetupRouter 入参加 `statisticClient *client.Client`。
|
||
|
||
- [ ] **Step 13.3: 启动 gateway + 测试 7 路由**
|
||
|
||
```bash
|
||
go build ./gateway
|
||
go run ./gateway &
|
||
# 7 个 HTTP 调用
|
||
curl -H "Authorization: Bearer $TOKEN" 'http://localhost:8080/api/v1/dashboard/today-overview?star_id=1'
|
||
# ...
|
||
```
|
||
|
||
Expected: 7 个 HTTP 端点全部返回 `{code:200, data:...}`
|
||
|
||
- [ ] **Step 13.4: 跑性能测试（k6/wrk）**
|
||
|
||
```bash
|
||
# k6 脚本（spec §5.4 性能基准）
|
||
k6 run --vus 10 --duration 30s - << 'EOF'
|
||
import http from 'k6/http';
|
||
export default function() {
|
||
http.get('http://localhost:8080/api/v1/dashboard/today-overview?star_id=1', {
|
||
headers: { Authorization: 'Bearer $TOKEN' }
|
||
});
|
||
}
|
||
EOF
|
||
```
|
||
|
||
Expected: P99 < 200ms（缓存命中时 <10ms）
|
||
|
||
- [ ] **Step 13.5: Commit**
|
||
|
||
```bash
|
||
git add backend/gateway/
|
||
git commit -m "feat(statistic): gateway 7 dashboard routes + statistic controller (P3 T13)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 14: 启动预热 + 健康检查 + 完整 E2E
|
||
|
||
**Files:**
|
||
- Modify: `backend/services/statisticService/main.go`
|
||
|
||
- [ ] **Step 14.1: 加预热逻辑**
|
||
|
||
在 main.go 启动 Dubbo server 后加：
|
||
|
||
```go
|
||
go func() {
|
||
time.Sleep(10 * time.Second) // 等服务起来
|
||
sampleUserID := int64(1) // 实际从配置读
|
||
for _, starID := range getTopNStars(config.PartitionCfg.PreCreateDays * 5) { // 35 个 top star
|
||
for _, rpc := range []string{"today_overview", "7day_income_curve"} { // 7 个 RPC 简写
|
||
go callAndWarmupCache(rpc, starID, sampleUserID, dashSvc)
|
||
}
|
||
}
|
||
}()
|
||
```
|
||
|
||
> 实际 cache warmup 函数封装 cache.Get / cache.Set。简化版只预热前 2 个 RPC（最常访问）。
|
||
|
||
- [ ] **Step 14.2: 跑 E2E 测试**
|
||
|
||
用 Playwright 跑前端 dashboard.vue 7 个数据点，加载到渲染不报错。
|
||
|
||
Expected: 7 个图表都能显示数据
|
||
|
||
- [ ] **Step 14.3: 切流量**
|
||
|
||
将前端 `USE_MOCK_API = false`，前端 `dashboardApi` 自动调真实后端。
|
||
|
||
- [ ] **Step 14.4: 跑契约测试**
|
||
|
||
```bash
|
||
# 用 proto 生成的 json schema 校验响应字段
|
||
go test ./integration/ -v -run TestContract
|
||
```
|
||
|
||
Expected: 7 个 RPC 响应字段不变
|
||
|
||
- [ ] **Step 14.5: Commit**
|
||
|
||
```bash
|
||
git add backend/services/statisticService/main.go backend/integration/
|
||
git commit -m "feat(statistic): P3 complete - 7 dashboard RPC end-to-end + cache warmup (T14)"
|
||
```
|
||
|
||
---
|
||
|
||
## P4 业务侧补全
|
||
|
||
### Task 15: galleryService + taskService 集成 TrackEvent
|
||
|
||
**Files:**
|
||
- Modify: `backend/services/galleryService/service/exhibition_service.go`
|
||
- Modify: `backend/services/taskService/service/revenue_service.go`
|
||
- Create: `backend/services/galleryService/.../exhibition_service_test.go`（mock TrackEvent）
|
||
- Create: `backend/services/taskService/.../revenue_service_test.go`
|
||
|
||
- [ ] **Step 15.1: galleryService 集成**
|
||
|
||
修改 `exhibition_service.go`：
|
||
|
||
```go
|
||
// 在 PlaceAsset 方法末尾
|
||
statistic.Get().TrackEvent(context.Background(), &eventPb.Event{
|
||
EventId: uuid.New().String(),
|
||
UserId: userID, StarId: starID,
|
||
EventType: "exhibition.start",
|
||
OccurredAt: time.Now().UnixMilli(),
|
||
Properties: map[string]string{"asset_id": strconv.FormatInt(assetID, 10), "slot_id": strconv.FormatInt(slotID, 10)},
|
||
})
|
||
|
||
// 在 RemoveFromSlot 方法末尾
|
||
statistic.Get().TrackEvent(context.Background(), &eventPb.Event{
|
||
EventId: uuid.New().String(),
|
||
UserId: userID, StarId: starID,
|
||
EventType: "exhibition.end",
|
||
OccurredAt: time.Now().UnixMilli(),
|
||
Properties: map[string]string{"asset_id": ..., "duration_ms": strconv.FormatInt(duration.Milliseconds(), 10)},
|
||
})
|
||
```
|
||
|
||
在 galleryService main.go 加 `statistic.Init(statisticClient)`。
|
||
|
||
- [ ] **Step 15.2: taskService 集成**
|
||
|
||
修改 `revenue_service.go` 的 `OnExhibitionCompleted` 方法末尾：
|
||
|
||
```go
|
||
statistic.Get().TrackEvent(context.Background(), &eventPb.Event{
|
||
EventId: uuid.New().String(),
|
||
UserId: userID, StarId: starID,
|
||
EventType: "exhibition.revenue",
|
||
OccurredAt: time.Now().UnixMilli(),
|
||
Properties: map[string]string{
|
||
"asset_id": strconv.FormatInt(assetID, 10),
|
||
"amount": strconv.FormatInt(amount, 10),
|
||
"duration_ms": strconv.FormatInt(duration.Milliseconds(), 10),
|
||
},
|
||
})
|
||
```
|
||
|
||
- [ ] **Step 15.3: 写 mock 测试**
|
||
|
||
`galleryService/.../exhibition_service_test.go`：
|
||
|
||
```go
|
||
// 用 mock 替换 statistic.Get() 验证 TrackEvent 被调用
|
||
func TestPlaceAsset_TrackEvent(t *testing.T) {
|
||
captured := &capturedEvent{}
|
||
statistic.SetMockForTest(captured) // 在 pkg/statistic 加测试用 mock 钩子
|
||
// ... 调 PlaceAsset ...
|
||
if captured.event == nil || captured.event.EventType != "exhibition.start" {
|
||
t.Fatal("TrackEvent not called with exhibition.start")
|
||
}
|
||
}
|
||
```
|
||
|
||
> 需要在 `pkg/statistic/client.go` 加 `SetMockForTest(c Capturer)` 函数。
|
||
|
||
- [ ] **Step 15.4: 启动两端 + 验证事件落库**
|
||
|
||
```bash
|
||
# 1. 启动 statisticService
|
||
# 2. 启动 galleryService / taskService
|
||
# 3. 触发 PlaceAsset / OnExhibitionCompleted
|
||
# 4. 查 statistic.events 看 exhibition.start / exhibition.end / exhibition.revenue
|
||
psql -c "SELECT event_type, COUNT(*) FROM statistic.events WHERE event_type IN ('exhibition.start','exhibition.end','exhibition.revenue') GROUP BY event_type"
|
||
```
|
||
|
||
Expected: 3 个 event_type 都有新行
|
||
|
||
- [ ] **Step 15.5: Commit**
|
||
|
||
```bash
|
||
git add backend/services/galleryService/ backend/services/taskService/ backend/pkg/statistic/
|
||
git commit -m "feat(statistic): P4 step 1 - galleryService + taskService TrackEvent integration (T15)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 16: assetService + userService 集成 TrackEvent + 看板数据 7 事件类型全出现
|
||
|
||
**Files:**
|
||
- Modify: `backend/services/assetService/service/mint_service.go`
|
||
- Modify: `backend/services/assetService/service/asset_level_service.go`
|
||
- Modify: `backend/services/userService/service/user_service.go`
|
||
- Tests
|
||
|
||
- [ ] **Step 16.1: assetService 集成（铸造）**
|
||
|
||
`mint_service.go` 的 `CreateMintOrder` 成功后：
|
||
|
||
```go
|
||
statistic.Get().TrackEvent(context.Background(), &eventPb.Event{
|
||
EventId: uuid.New().String(),
|
||
UserId: userID, StarId: starID,
|
||
EventType: "asset.mint",
|
||
OccurredAt: time.Now().UnixMilli(),
|
||
Properties: map[string]string{"asset_id": strconv.FormatInt(assetID, 10), "level": level},
|
||
})
|
||
```
|
||
|
||
- [ ] **Step 16.2: assetService 集成（升级）**
|
||
|
||
`asset_level_service.go` 的 `CheckUpgrade` 返回 true 触发升级后（或 `logLevelChange` 调用前）：
|
||
|
||
```go
|
||
statistic.Get().TrackEvent(context.Background(), &eventPb.Event{
|
||
EventId: uuid.New().String(),
|
||
UserId: userID, StarId: starID,
|
||
EventType: "asset.level_up",
|
||
OccurredAt: time.Now().UnixMilli(),
|
||
Properties: map[string]string{
|
||
"asset_id": strconv.FormatInt(assetID, 10),
|
||
"from": fromLevel, "to": toLevel,
|
||
},
|
||
})
|
||
```
|
||
|
||
> **P1 末向 assetService 同学确认实际升级触发点**（CheckUpgrade 返回 true 后在哪行调升级），用真实代码位置。
|
||
|
||
- [ ] **Step 16.3: userService 集成（水晶账本变动）**
|
||
|
||
`user_service.go` 的 `UpdateCrystalBalance` 方法末尾（成功后）：
|
||
|
||
```go
|
||
statistic.Get().TrackEvent(context.Background(), &eventPb.Event{
|
||
EventId: uuid.New().String(),
|
||
UserId: req.UserId, StarId: req.StarId,
|
||
EventType: "crystal.change",
|
||
OccurredAt: time.Now().UnixMilli(),
|
||
Properties: map[string]string{
|
||
"amount": strconv.FormatInt(req.Delta, 10),
|
||
"reason": req.Reason, // 业务侧传
|
||
},
|
||
})
|
||
```
|
||
|
||
- [ ] **Step 16.4: 跑全量联调**
|
||
|
||
```bash
|
||
# 触发 7 个事件类型各 1 次
|
||
# 用项目现有 e2e 脚本或 API
|
||
# 然后查 events 表
|
||
psql -c "SELECT event_type, COUNT(*) FROM statistic.events WHERE received_at > NOW() - INTERVAL '1 hour' GROUP BY event_type ORDER BY event_type"
|
||
```
|
||
|
||
Expected: 7 个 event_type 都有新行（asset.like / asset.mint / exhibition.start / exhibition.end / exhibition.revenue / asset.level_up / crystal.change）
|
||
|
||
- [ ] **Step 16.5: 验证看板 7 RPC 数据完整性**
|
||
|
||
```bash
|
||
# 切前端 USE_MOCK_API=false
|
||
# 登录 → 进入 dashboard 页 → 检查 7 个图表都有数据
|
||
```
|
||
|
||
Expected: 7 个图表都有真实数据，无空状态
|
||
|
||
- [ ] **Step 16.6: 跑所有 P4 服务单测**
|
||
|
||
```bash
|
||
cd backend
|
||
go test ./services/galleryService/... ./services/taskService/... ./services/assetService/... ./services/userService/... -v
|
||
```
|
||
|
||
Expected: 全部通过
|
||
|
||
- [ ] **Step 16.7: Commit**
|
||
|
||
```bash
|
||
git add backend/services/assetService/ backend/services/userService/
|
||
git commit -m "feat(statistic): P4 complete - assetService + userService TrackEvent + 7 event types end-to-end (T16)"
|
||
```
|
||
|
||
---
|
||
|
||
## 完成验证（本期交付清单）
|
||
|
||
### 功能
|
||
|
||
- [ ] 7 个看板 RPC 端到端通（gateway → statisticService → MV）
|
||
- [ ] `week_rank` 完整实现（排名准确）
|
||
- [ ] TrackEvent 接收 + 去重 + 批量
|
||
- [ ] 4 MV + 3 metric 全部自动刷新
|
||
- [ ] events 按日分区自动管理（7 天预创建 + 30 天清理）
|
||
- [ ] 5 个业务服务全部集成 TrackEvent
|
||
- [ ] 7 个事件类型（asset.like / asset.mint / exhibition.start / end / revenue / level_up / crystal.change）全部接入
|
||
|
||
### 性能
|
||
|
||
- [ ] 看板单 RPC 缓存命中 P99 < 10ms
|
||
- [ ] 看板 7 RPC 并发 P99 < 500ms
|
||
- [ ] TrackEvent P99 < 50ms
|
||
- [ ] 物化视图刷新 < 30s/视图
|
||
- [ ] 缓存命中率 > 80%（启动预热后）
|
||
|
||
### 可靠性
|
||
|
||
- [ ] 单元测试覆盖率 > 80%
|
||
- [ ] 集成测试覆盖 100% RPC（dockertest 真实 PG）
|
||
- [ ] TrackEvent 失败业务方有 retry（业务侧 defer + recover）
|
||
- [ ] partition 缺失自动创建
|
||
- [ ] MV 刷新失败有 refresh_log 记录
|
||
|
||
### 可观测性
|
||
|
||
- [ ] Prometheus 指标 20+ 全部埋点（dashboard_rpc / event_track / mv_refresh / worker_running / events_partition_count）
|
||
- [ ] /healthz 端点 200
|
||
- [ ] refresh_log 表完整记录
|
||
- [ ] /metrics 端点暴露
|
||
|
||
### 集成
|
||
|
||
- [ ] Gateway 7 路由注册成功
|
||
- [ ] 前端 USE_MOCK_API=false 切流量成功
|
||
- [ ] 业务侧 5 个服务部署后事件能落库
|
||
- [ ] statistic schema 10 张表全部存在
|
||
|
||
### 安全
|
||
|
||
- [ ] JWT 鉴权在 gateway 拦截
|
||
- [ ] 粉丝身份校验（userService.fan_profile 调通）
|
||
- [ ] JSONB 注入防护（参数化 SQL）
|
||
- [ ] 限流配置生效（spec §4.6）
|
||
|
||
---
|
||
|
||
## 自我审查
|
||
|
||
### 1. Spec 覆盖
|
||
|
||
| Spec 项 | 任务 |
|
||
|---------|------|
|
||
| 看板 7 RPC 实现 | T11, T12, T13 |
|
||
| 事件采集 (TrackEvent/BatchTrackEvent) | T4, T5, T6, T9 |
|
||
| 预聚表 (3 张) | T6 (recent), T7 (weekly, upcoming) |
|
||
| 物化视图 (4 个) | T3 (DDL), T10 (刷新) |
|
||
| 分区管理 | T3 (DDL), T8 (worker) |
|
||
| Gateway 7 路由 + JWT + 粉丝身份 | T13 |
|
||
| 5 业务服务集成 | T9 (social), T15 (gallery, task), T16 (asset, user) |
|
||
| Prometheus 指标 | T3 (骨架), T9 (事件), T10 (MV), T12 (看板) |
|
||
| healthz | T3 |
|
||
| 启动预热 | T14 |
|
||
| P1 末预检查 | T3 step 3.9 |
|
||
|
||
### 2. 占位符扫描
|
||
|
||
- ❌ 无 TBD/TODO
|
||
- ❌ 无 "implement later" / "fill in details"
|
||
- ❌ 无 "add appropriate error handling"（错误处理在每个任务的具体代码中已写）
|
||
- ❌ 无 "Similar to Task N"（每个任务代码独立完整）
|
||
- ✅ 所有代码步骤包含完整代码
|
||
|
||
### 3. 类型一致性
|
||
|
||
| 任务 | 定义 | 后续使用 |
|
||
|------|------|----------|
|
||
| T4 | `Event` model | T5 repo 用, T5 service 用, T6 worker 用, T9 provider 用 |
|
||
| T4 | `EventSink` 接口 | T4 实现, T5 service 注入, T9 main 注入 |
|
||
| T5 | `EventRepository.InsertBatch` | T6 flusher 调, T9 main 注入 |
|
||
| T5 | `EventService.TrackEvent` | T9 provider 调 |
|
||
| T6 | `MetricRepository.UpsertRecentLevelUp` | T6 flusher 调, T9 main 注入 |
|
||
| T7 | `MetricRepository.RefreshWeeklyUserIncome` | T7 worker 调, T9 main 注入 |
|
||
| T7 | `MetricRepository.RefreshUpcomingLevelUps` | T7 worker 调, T9 main 注入 |
|
||
| T8 | `Partitioner.EnsureFuturePartitions` | T8 Start 调, T9 main 注入 |
|
||
| T9 | `StatisticInternalProvider` | T9 main 注册 Dubbo |
|
||
| T10 | `Materializer.RefreshOne` | T10 Start 调 |
|
||
| T11 | `DashboardRepository.GetWeekRank` | T11 service 调 |
|
||
| T11 | `DashboardService.GetTodayOverview` | T12 mobile provider 调 |
|
||
| T12 | `StatisticMobileProvider` | T13 gateway controller 调 |
|
||
| T13 | `StatisticController.GetTodayOverview` | T14 E2E 测试 |
|
||
| T9 | `pkg/statistic.Client.TrackEvent` | T9 socialService 调, T15 gallery/task 调, T16 asset/user 调 |
|
||
|
||
所有类型/方法在定义任务和后续使用任务间一致。
|
||
|
||
---
|
||
|
||
**计划完成，保存到 `docs/superpowers/plans/2026-06-08-statistic-kanban-and-event-implementation.md`。**
|