数据采集架构设计

package analytics

import (
    "context"
    "encoding/json"
    "log"
    "time"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/metric"
)

type GameEventCollector struct {
    tracer otel.Tracer
    meter  otel.Meter

    // 核心指标
    userLoginCounter     metric.Int64Counter
    levelCompleteCounter metric.Int64Counter
    currencyEarnCounter  metric.Int64Counter
    currencySpendCounter metric.Int64Counter
}

func NewGameEventCollector() *GameEventCollector {
    tracer := otel.Tracer("game.server")
    meter := otel.Meter("game.server")

    return &GameEventCollector{
        tracer: tracer,
        meter:  meter,
        userLoginCounter: meter.Int64Counter("game.user.login.total"),
        levelCompleteCounter: meter.Int64Counter("game.level.complete.total"),
        currencyEarnCounter: meter.Int64Counter("game.currency.earn.total"),
        currencySpendCounter: meter.Int64Counter("game.currency.spend.total"),
    }
}

// 用户登录事件
func (g *GameEventCollector) TrackUserLogin(ctx context.Context, req UserLoginRequest) {
    ctx, span := g.tracer.Start(ctx, "user.login")
    defer span.End()

    span.SetAttributes(
        attribute.String("user.id", req.UserID),
        attribute.String("login.method", req.Method),
        attribute.String("client.ip", req.ClientIP),
        attribute.String("device.platform", req.Platform),
    )

    g.userLoginCounter.Add(ctx, 1, metric.WithAttributes(
        attribute.String("game.id", req.GameID),
        attribute.String("game.env", req.Environment),
        attribute.String("login.method", req.Method),
    ))

    // 发送到消息队列
    event := GameEvent{
        EventType: "user.login",
        UserID:    req.UserID,
        GameID:    req.GameID,
        Timestamp: time.Now(),
        Properties: map[string]interface{}{
            "login_method": req.Method,
            "client_ip":   req.ClientIP,
            "platform":    req.Platform,
            "server_time": time.Now().Unix(),
        },
    }

    g.publishEvent(ctx, event)
}

// 关卡完成事件
func (g *GameEventCollector) TrackLevelComplete(ctx context.Context, req LevelCompleteRequest) {
    ctx, span := g.tracer.Start(ctx, "gameplay.level.complete")
    defer span.End()

    completionRate := 0.0
    if req.Success {
        completionRate = 1.0
    }

    span.SetAttributes(
        attribute.String("user.id", req.UserID),
        attribute.String("level.id", req.LevelID),
        attribute.Bool("level.success", req.Success),
        attribute.Float64("level.duration", req.Duration.Seconds()),
        attribute.Int("level.attempts", req.Attempts),
    )

    g.levelCompleteCounter.Add(ctx, 1, metric.WithAttributes(
        attribute.String("level.id", req.LevelID),
        attribute.Bool("success", req.Success),
    ))

    // 服务器权威数据验证
    if err := g.validateLevelCompletion(ctx, req); err != nil {
        log.Printf("关卡完成数据异常: %v", err)
        span.SetAttributes(attribute.Bool("validation.failed", true))
        return
    }

    event := GameEvent{
        EventType: "gameplay.level.complete",
        UserID:    req.UserID,
        GameID:    req.GameID,
        Timestamp: time.Now(),
        Properties: map[string]interface{}{
            "level_id":        req.LevelID,
            "success":         req.Success,
            "duration":        req.Duration.Seconds(),
            "attempts":        req.Attempts,
            "completion_rate": completionRate,
            "score":          req.Score,
            "server_validated": true,
        },
    }

    g.publishEvent(ctx, event)
}

// 经济系统事件
func (g *GameEventCollector) TrackEconomyTransaction(ctx context.Context, transaction EconomyTransaction) {
    ctx, span := g.tracer.Start(ctx, "economy.transaction")
    defer span.End()

    span.SetAttributes(
        attribute.String("user.id", transaction.UserID),
        attribute.String("transaction.type", transaction.Type),
        attribute.String("currency.type", transaction.CurrencyType),
        attribute.Int64("amount", transaction.Amount),
        attribute.String("source", transaction.Source),
    )

    if transaction.Type == "earn" {
        g.currencyEarnCounter.Add(ctx, transaction.Amount, metric.WithAttributes(
            attribute.String("currency.type", transaction.CurrencyType),
            attribute.String("source", transaction.Source),
        ))
    } else {
        g.currencySpendCounter.Add(ctx, transaction.Amount, metric.WithAttributes(
            attribute.String("currency.type", transaction.CurrencyType),
            attribute.String("purpose", transaction.Purpose),
        ))
    }

    event := GameEvent{
        EventType: fmt.Sprintf("economy.%s", transaction.Type),
        UserID:    transaction.UserID,
        GameID:    transaction.GameID,
        Timestamp: time.Now(),
        Properties: map[string]interface{}{
            "currency_type":    transaction.CurrencyType,
            "amount":          transaction.Amount,
            "source":          transaction.Source,
            "purpose":         transaction.Purpose,
            "balance_before":  transaction.BalanceBefore,
            "balance_after":   transaction.BalanceAfter,
            "server_authoritative": true,
        },
    }

    g.publishEvent(ctx, event)
}

// 数据校验逻辑
func (g *GameEventCollector) validateLevelCompletion(ctx context.Context, req LevelCompleteRequest) error {
    // 校验关卡是否存在
    if !g.levelExists(req.LevelID) {
        return fmt.Errorf("关卡不存在: %s", req.LevelID)
    }

    // 校验完成时间合理性
    minDuration := g.getLevelMinDuration(req.LevelID)
    if req.Success && req.Duration < minDuration {
        return fmt.Errorf("完成时间异常: %v < %v", req.Duration, minDuration)
    }

    // 校验分数合理性
    maxScore := g.getLevelMaxScore(req.LevelID)
    if req.Score > maxScore {
        return fmt.Errorf("分数异常: %d > %d", req.Score, maxScore)
    }

    return nil
}

type PaymentEventCollector struct {
    tracer otel.Tracer
    meter  otel.Meter

    revenueCounter     metric.Float64Counter
    transactionCounter metric.Int64Counter
}

func (p *PaymentEventCollector) TrackPayment(ctx context.Context, payment PaymentEvent) {
    ctx, span := p.tracer.Start(ctx, "economy.payment")
    defer span.End()

    span.SetAttributes(
        attribute.String("user.id", payment.UserID),
        attribute.String("order.id", payment.OrderID),
        attribute.String("product.id", payment.ProductID),
        attribute.Float64("amount", payment.Amount),
        attribute.String("currency", payment.Currency),
        attribute.String("status", payment.Status),
        attribute.String("payment.method", payment.PaymentMethod),
    )

    // 只有成功支付才计入收入
    if payment.Status == "success" {
        p.revenueCounter.Add(ctx, payment.Amount, metric.WithAttributes(
            attribute.String("product.id", payment.ProductID),
            attribute.String("currency", payment.Currency),
            attribute.String("payment.method", payment.PaymentMethod),
        ))
    }

    p.transactionCounter.Add(ctx, 1, metric.WithAttributes(
        attribute.String("status", payment.Status),
        attribute.String("payment.method", payment.PaymentMethod),
    ))

    event := GameEvent{
        EventType: "economy.payment",
        UserID:    payment.UserID,
        GameID:    payment.GameID,
        Timestamp: time.Now(),
        Properties: map[string]interface{}{
            "order_id":       payment.OrderID,
            "product_id":     payment.ProductID,
            "amount":         payment.Amount,
            "currency":       payment.Currency,
            "status":         payment.Status,
            "payment_method": payment.PaymentMethod,
            "platform_fee":   payment.PlatformFee,
            "net_revenue":    payment.Amount - payment.PlatformFee,
            "verified":       true,
        },
    }

    p.publishEvent(ctx, event)
}

// 实时指标计算
type RealTimeAggregator struct {
    redis    *redis.Client
    interval time.Duration
}

func (r *RealTimeAggregator) UpdateDAU(ctx context.Context, gameID, env, userID string) {
    key := fmt.Sprintf("dau:%s:%s:%s", gameID, env, time.Now().Format("2006-01-02"))

    // 使用HyperLogLog进行去重统计
    r.redis.PFAdd(ctx, key, userID)
    r.redis.Expire(ctx, key, 48*time.Hour) // 保留2天
}

func (r *RealTimeAggregator) GetCurrentDAU(ctx context.Context, gameID, env string) (int64, error) {
    key := fmt.Sprintf("dau:%s:%s:%s", gameID, env, time.Now().Format("2006-01-02"))
    return r.redis.PFCount(ctx, key).Result()
}

func (r *RealTimeAggregator) UpdateRevenue(ctx context.Context, gameID, env string, amount float64) {
    key := fmt.Sprintf("revenue:%s:%s:%s", gameID, env, time.Now().Format("2006-01-02"))

    // 累计收入
    r.redis.IncrByFloat(ctx, key, amount)
    r.redis.Expire(ctx, key, 7*24*time.Hour) // 保留7天
}

type HybridEventValidator struct {
    clientEvents map[string]ClientEvent
    serverEvents map[string]ServerEvent
    tolerance    time.Duration // 时间容忍度
}

func (h *HybridEventValidator) ValidateEvent(clientEvent ClientEvent) ValidationResult {
    // 查找对应的服务器事件
    serverEvent, exists := h.serverEvents[clientEvent.TransactionID]

    if !exists {
        return ValidationResult{
            Status: "server_missing",
            Action: "discard", // 丢弃客户端事件
            Reason: "服务器无对应事件",
        }
    }

    // 时间差验证
    timeDiff := abs(clientEvent.Timestamp - serverEvent.Timestamp)
    if timeDiff > h.tolerance {
        return ValidationResult{
            Status: "time_mismatch",
            Action: "flag_suspicious",
            Reason: fmt.Sprintf("时间差异过大: %v", timeDiff),
        }
    }

    // 关键数据一致性验证
    if clientEvent.UserID != serverEvent.UserID {
        return ValidationResult{
            Status: "user_mismatch",
            Action: "discard",
            Reason: "用户ID不匹配",
        }
    }

    // 业务逻辑验证
    if err := h.validateBusinessLogic(clientEvent, serverEvent); err != nil {
        return ValidationResult{
            Status: "business_logic_error",
            Action: "use_server_data",
            Reason: err.Error(),
        }
    }

    return ValidationResult{
        Status: "validated",
        Action: "merge",
        Reason: "验证通过",
    }
}

func (h *HybridEventValidator) MergeEvents(clientEvent ClientEvent, serverEvent ServerEvent) MergedEvent {
    return MergedEvent{
        // 使用服务器权威数据
        UserID:    serverEvent.UserID,
        EventType: serverEvent.EventType,
        Timestamp: serverEvent.Timestamp,

        // 合并业务数据
        GameData: serverEvent.GameData,

        // 添加客户端体验数据
        ClientMetrics: ClientMetrics{
            FPS:          clientEvent.FPS,
            LoadTime:     clientEvent.LoadTime,
            NetworkRTT:   clientEvent.NetworkRTT,
            BatteryLevel: clientEvent.BatteryLevel,
        },

        // 标记数据来源
        DataSources: []string{"client", "server"},
        Validated:   true,
    }
}

type EventRouter struct {
    rules map[string]RoutingRule
}

type RoutingRule struct {
    DataSource   string        // "client", "server", "hybrid"
    Importance   int           // 1-5，重要性等级
    SamplingRate float64       // 0-1，采样率
    QualityCheck bool          // 是否需要质量检查
    RealTime     bool          // 是否需要实时处理
}

func (e *EventRouter) RouteEvent(event BaseEvent) RoutingDecision {
    rule, exists := e.rules[event.EventType]
    if !exists {
        rule = e.getDefaultRule()
    }

    // 根据重要性调整采样率
    samplingRate := rule.SamplingRate
    if event.UserVIPLevel > 0 {
        samplingRate = math.Min(1.0, samplingRate * 2) // VIP用户提高采样率
    }

    // 随机采样决策
    if rand.Float64() > samplingRate {
        return RoutingDecision{
            Action: "discard",
            Reason: "sampling_filtered",
        }
    }

    return RoutingDecision{
        Action:       "process",
        DataSource:   rule.DataSource,
        QualityCheck: rule.QualityCheck,
        RealTime:     rule.RealTime,
        Priority:     rule.Importance,
    }
}

// 事件重要性规则配置
var defaultEventRules = map[string]RoutingRule{
    "user.login":           {DataSource: "server", Importance: 5, SamplingRate: 1.0, RealTime: true},
    "user.register":        {DataSource: "server", Importance: 5, SamplingRate: 1.0, RealTime: true},
    "economy.payment":      {DataSource: "server", Importance: 5, SamplingRate: 1.0, QualityCheck: true},
    "gameplay.level.complete": {DataSource: "hybrid", Importance: 4, SamplingRate: 1.0, RealTime: true},
    "ui.button_click":      {DataSource: "client", Importance: 2, SamplingRate: 0.1, RealTime: false},
    "performance.sample":   {DataSource: "client", Importance: 3, SamplingRate: 0.05, RealTime: false},
}

type DataCollectionGateway struct {
    rateLimiter  *rate.Limiter
    validator    *EventValidator
    router       *EventRouter
    queue        MessageQueue
}

func (d *DataCollectionGateway) ProcessEvent(ctx context.Context, rawEvent RawEvent) error {
    // 1. 频率限制
    if !d.rateLimiter.Allow() {
        return errors.New("rate limit exceeded")
    }

    // 2. 数据验证
    validationResult := d.validator.Validate(rawEvent)
    if !validationResult.Valid {
        return fmt.Errorf("validation failed: %s", validationResult.Error)
    }

    // 3. 事件路由
    routingDecision := d.router.RouteEvent(rawEvent)
    if routingDecision.Action == "discard" {
        return nil // 正常丢弃，不算错误
    }

    // 4. 数据丰富化
    enrichedEvent := d.enrichEvent(rawEvent)

    // 5. 发送到消息队列
    return d.queue.Publish(ctx, enrichedEvent, QueueOptions{
        Priority:  routingDecision.Priority,
        RealTime:  routingDecision.RealTime,
        Retry:     3,
    })
}

func (d *DataCollectionGateway) enrichEvent(rawEvent RawEvent) EnrichedEvent {
    enrichedEvent := EnrichedEvent{
        RawEvent:  rawEvent,
        Timestamp: time.Now(),
        GatewayID: "gateway-001",
    }

    // IP地理位置解析
    if rawEvent.ClientIP != "" {
        geoInfo := d.geoResolver.Resolve(rawEvent.ClientIP)
        enrichedEvent.GeoInfo = geoInfo
    }

    // 用户画像数据补充
    if rawEvent.UserID != "" {
        userProfile := d.userProfileService.GetProfile(rawEvent.UserID)
        enrichedEvent.UserProfile = userProfile
    }

    // 设备指纹分析
    if rawEvent.DeviceInfo != nil {
        deviceFingerprint := d.deviceAnalyzer.Analyze(rawEvent.DeviceInfo)
        enrichedEvent.DeviceFingerprint = deviceFingerprint
    }

    return enrichedEvent
}

type ConsistencyChecker struct {
    clientDataWindow time.Duration
    serverDataWindow time.Duration
    tolerance        float64
}

func (c *ConsistencyChecker) CheckDailyConsistency(date time.Time) ConsistencyReport {
    // 获取客户端和服务器的DAU数据
    clientDAU := c.getClientDAU(date)
    serverDAU := c.getServerDAU(date)

    // 计算差异百分比
    diff := math.Abs(float64(clientDAU-serverDAU)) / float64(serverDAU)

    report := ConsistencyReport{
        Date:      date,
        ClientDAU: clientDAU,
        ServerDAU: serverDAU,
        Difference: diff,
        Status:    "normal",
    }

    if diff > c.tolerance {
        report.Status = "anomaly"
        report.Alert = true
        report.Reason = fmt.Sprintf("DAU差异过大: %.2f%%", diff*100)
    }

    return report
}

type AnomalyDetector struct {
    models map[string]*TimeSeriesModel
}

func (a *AnomalyDetector) DetectAnomaly(metric string, value float64, timestamp time.Time) bool {
    model, exists := a.models[metric]
    if !exists {
        return false // 无基线数据
    }

    // 预测期望值
    expected := model.Predict(timestamp)

    // 计算标准差
    stdDev := model.GetStandardDeviation()

    // 3-sigma规则检测异常
    return math.Abs(value-expected) > 3*stdDev
}