133 lines
4.8 KiB
Go
133 lines
4.8 KiB
Go
package provider
|
||
|
||
import (
|
||
"context"
|
||
"strconv"
|
||
"time"
|
||
|
||
"dubbo.apache.org/dubbo-go/v3/common/constant"
|
||
"go.uber.org/zap"
|
||
|
||
"github.com/topfans/backend/pkg/logger"
|
||
pb "github.com/topfans/backend/pkg/proto/statistic"
|
||
"github.com/topfans/backend/services/statisticService/metrics"
|
||
"github.com/topfans/backend/services/statisticService/service"
|
||
)
|
||
|
||
// StatisticCombinedProvider 完整 StatisticService 实现（9 RPC）
|
||
// - 看板 7 RPC（T12 由 DashboardService 实现）
|
||
// - 事件 2 RPC（T9 由 StatisticInternalProvider 实现）
|
||
type StatisticCombinedProvider struct {
|
||
*StatisticInternalProvider
|
||
dashSvc *service.DashboardService
|
||
}
|
||
|
||
// NewStatisticCombinedProvider 构造
|
||
func NewStatisticCombinedProvider(internal *StatisticInternalProvider, dashSvc *service.DashboardService) *StatisticCombinedProvider {
|
||
return &StatisticCombinedProvider{
|
||
StatisticInternalProvider: internal,
|
||
dashSvc: dashSvc,
|
||
}
|
||
}
|
||
|
||
func userIDFromContext(ctx context.Context) int64 {
|
||
// 必须从 Dubbo attachments 读取（与 assetService/socialService/userService 一致）
|
||
// 协议：gateway 通过 ctx.Value(constant.AttachmentKey, map{"user_id": "1"}) 写入
|
||
// 经过 triple 协议传输后，gateway 的 string 在 attachments 里变成 ["1"]（单元素字符串数组）
|
||
// 之前用 ctx.Value("user_id") 直接读 string key，永远拿到 nil → 返回 0
|
||
if attachments := ctx.Value(constant.AttachmentKey); attachments != nil {
|
||
if attMap, ok := attachments.(map[string]interface{}); ok {
|
||
if v, ok := attMap["user_id"]; ok {
|
||
switch s := v.(type) {
|
||
case string:
|
||
n, _ := strconv.ParseInt(s, 10, 64)
|
||
return n
|
||
case int64:
|
||
return s
|
||
case int:
|
||
return int64(s)
|
||
case float64:
|
||
return int64(s)
|
||
case []string:
|
||
if len(s) > 0 {
|
||
n, _ := strconv.ParseInt(s[0], 10, 64)
|
||
return n
|
||
}
|
||
case []interface{}:
|
||
if len(s) > 0 {
|
||
switch inner := s[0].(type) {
|
||
case string:
|
||
n, _ := strconv.ParseInt(inner, 10, 64)
|
||
return n
|
||
case int64:
|
||
return inner
|
||
case int:
|
||
return int64(inner)
|
||
case float64:
|
||
return int64(inner)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func (p *StatisticCombinedProvider) 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())
|
||
}
|
||
|
||
// ===== 看板 7 RPC =====
|
||
|
||
func (p *StatisticCombinedProvider) GetTodayOverview(ctx context.Context, req *pb.GetTodayOverviewRequest) (*pb.GetTodayOverviewResponse, error) {
|
||
t0 := time.Now()
|
||
defer func() { p.recordRPC("GetTodayOverview", t0, nil) }()
|
||
resp, err := p.dashSvc.GetTodayOverview(ctx, userIDFromContext(ctx), req.StarId)
|
||
if err != nil {
|
||
logger.Logger.Warn("GetTodayOverview failed", zap.Error(err))
|
||
}
|
||
return resp, err
|
||
}
|
||
|
||
func (p *StatisticCombinedProvider) Get7DayIncomeCurve(ctx context.Context, req *pb.Get7DayIncomeCurveRequest) (*pb.Get7DayIncomeCurveResponse, error) {
|
||
t0 := time.Now()
|
||
defer func() { p.recordRPC("Get7DayIncomeCurve", t0, nil) }()
|
||
return p.dashSvc.Get7DayIncomeCurve(ctx, userIDFromContext(ctx), req.StarId)
|
||
}
|
||
|
||
func (p *StatisticCombinedProvider) GetExhibitionIncomeSummary(ctx context.Context, req *pb.GetExhibitionIncomeSummaryRequest) (*pb.GetExhibitionIncomeSummaryResponse, error) {
|
||
t0 := time.Now()
|
||
defer func() { p.recordRPC("GetExhibitionIncomeSummary", t0, nil) }()
|
||
return p.dashSvc.GetExhibitionIncomeSummary(ctx, userIDFromContext(ctx), req.StarId)
|
||
}
|
||
|
||
func (p *StatisticCombinedProvider) GetLikeIncomeByLevel(ctx context.Context, req *pb.GetLikeIncomeByLevelRequest) (*pb.GetLikeIncomeByLevelResponse, error) {
|
||
t0 := time.Now()
|
||
defer func() { p.recordRPC("GetLikeIncomeByLevel", t0, nil) }()
|
||
return p.dashSvc.GetLikeIncomeByLevel(ctx, userIDFromContext(ctx), req.StarId)
|
||
}
|
||
|
||
func (p *StatisticCombinedProvider) GetTopAssetsByEarning(ctx context.Context, req *pb.GetTopAssetsByEarningRequest) (*pb.GetTopAssetsByEarningResponse, error) {
|
||
t0 := time.Now()
|
||
defer func() { p.recordRPC("GetTopAssetsByEarning", t0, nil) }()
|
||
return p.dashSvc.GetTopAssetsByEarning(ctx, userIDFromContext(ctx), req.StarId)
|
||
}
|
||
|
||
func (p *StatisticCombinedProvider) GetAssetLevelDistribution(ctx context.Context, req *pb.GetAssetLevelDistributionRequest) (*pb.GetAssetLevelDistributionResponse, error) {
|
||
t0 := time.Now()
|
||
defer func() { p.recordRPC("GetAssetLevelDistribution", t0, nil) }()
|
||
return p.dashSvc.GetAssetLevelDistribution(ctx, userIDFromContext(ctx), req.StarId)
|
||
}
|
||
|
||
func (p *StatisticCombinedProvider) GetAssetUpgradeProgress(ctx context.Context, req *pb.GetAssetUpgradeProgressRequest) (*pb.GetAssetUpgradeProgressResponse, error) {
|
||
t0 := time.Now()
|
||
defer func() { p.recordRPC("GetAssetUpgradeProgress", t0, nil) }()
|
||
return p.dashSvc.GetAssetUpgradeProgress(ctx, userIDFromContext(ctx), req.StarId)
|
||
}
|