flowchart TB
    subgraph DataSource["数据源"]
        Kafka["Kafka<br/>事件流"]
        HTTPPush["HTTP API<br/>主动推送"]
        LogCollector["日志采集"]
    end

    subgraph AnalyticsSvc["Analytics Service"]
        subgraph ServerLayer["Server层"]
            HTTPServer["HTTP Server<br/>Gin路由"]
        end

        subgraph BizLayer["Biz层（业务逻辑）"]
            MetricUsecase["MetricUsecase<br/>指标用例"]
            ReportUsecase["ReportUsecase<br/>报表用例"]
            DashboardUsecase["DashboardUsecase<br/>仪表盘用例"]
        end

        subgraph DataLayer["Data层（数据访问）"]
            MetricRepo["MetricRepository<br/>指标仓储"]
            ReportRepo["ReportRepository<br/>报表仓储"]
            CHClient["ClickHouseClient<br/>CH客户端"]
            CacheClient["CacheClient<br/>缓存客户端"]
        end
    end

    subgraph Storage["存储层"]
        ClickHouse["ClickHouse<br/>分析数据库"]
        PostgreSQL["PostgreSQL<br/>元数据/报表"]
        Redis["Redis<br/>实时缓存"]
    end

    subgraph Consumers["事件消费者"]
        MessageConsumer["MessageConsumer<br/>消息事件"]
        DocumentConsumer["DocumentConsumer<br/>文档事件"]
        AIUsageConsumer["AIUsageConsumer<br/>AI使用事件"]
    end

    subgraph Aggregation["数据聚合"]
        Materialized["Materialized Views<br/>物化视图"]
        RealtimeAgg["RealtimeAggregator<br/>实时聚合器"]
    end

    DataSource --> Consumers
    HTTPPush --> HTTPServer
    Consumers --> ClickHouse

    HTTPServer --> MetricUsecase
    HTTPServer --> ReportUsecase
    HTTPServer --> DashboardUsecase

    MetricUsecase --> MetricRepo
    ReportUsecase --> ReportRepo
    ReportUsecase --> MetricRepo
    DashboardUsecase --> CHClient
    DashboardUsecase --> CacheClient

    MetricRepo --> CHClient
    MetricRepo --> CacheClient
    ReportRepo --> PostgreSQL
    CHClient --> ClickHouse
    CacheClient --> Redis

    ClickHouse --> Materialized
    Materialized --> RealtimeAgg
    RealtimeAgg --> Redis

    style DataSource fill:#e3f2fd
    style ServerLayer fill:#fff3e0
    style BizLayer fill:#ffe0b2
    style DataLayer fill:#ffccbc
    style Storage fill:#e0f2f1
    style Consumers fill:#f3e5f5
    style Aggregation fill:#fff9c4

{
  "event_type": "message.created",
  "tenant_id": "tenant_abc",
  "user_id": "user_123",
  "conversation_id": "conv_456",
  "role": "assistant",
  "tokens": 250,
  "model": "gpt-4",
  "cost_usd": 0.0075,
  "timestamp": "2025-01-27T10:00:00Z"
}

{
  "event_type": "document.indexed",
  "tenant_id": "tenant_abc",
  "document_id": "doc_789",
  "chunks_count": 150,
  "processing_time_ms": 5000,
  "timestamp": "2025-01-27T10:00:00Z"
}

{
  "event_type": "ai.model.invoked",
  "tenant_id": "tenant_abc",
  "model_name": "gpt-4",
  "provider": "openai",
  "tokens": 1000,
  "latency_ms": 1500,
  "cost_usd": 0.03,
  "timestamp": "2025-01-27T10:00:00Z"
}

CREATE MATERIALIZED VIEW metrics_hourly_mv
TO metrics_hourly
AS SELECT
    toStartOfHour(timestamp) as hour,
    tenant_id,
    countIf(event_type = 'message.created') as message_count,
    sumIf(tokens, event_type = 'message.created') as total_tokens,
    sumIf(cost_usd, event_type = 'message.created') as total_cost,
    uniq(user_id) as active_users
FROM events
GROUP BY hour, tenant_id;

classDiagram
    class Metric {
        +string ID
        +string TenantID
        +MetricType Type
        +string Name
        +float64 Value
        +map Dimensions
        +time.Time Timestamp
    }

    class UsageStats {
        +string TenantID
        +int64 TotalConversations
        +int64 TotalMessages
        +int64 TotalTokens
        +float64 TotalCost
        +int64 ActiveUsers
        +TimePeriod Period
        +time.Time StartTime
        +time.Time EndTime
    }

    class ModelStats {
        +string TenantID
        +string ModelName
        +string Provider
        +int64 RequestCount
        +int64 TotalTokens
        +float64 TotalCost
        +float64 AvgLatency
        +float64 ErrorRate
        +TimePeriod Period
    }

    class UserBehavior {
        +string TenantID
        +string UserID
        +int64 ConversationCount
        +int64 MessageCount
        +float64 AvgSessionDuration
        +time.Time LastActiveTime
        +TimePeriod Period
    }

    class RealtimeStats {
        +string TenantID
        +float64 CurrentQPS
        +int64 CurrentActiveUsers
        +float64 CurrentLatency
        +time.Time Timestamp
    }

    class CostBreakdown {
        +string TenantID
        +float64 TotalCost
        +CostByModel map
        +CostByFeature map
        +TimePeriod Period
    }

    class Report {
        +string ID
        +string TenantID
        +ReportType Type
        +string Name
        +ReportStatus Status
        +string FileURL
        +map Data
        +string CreatedBy
        +time.Time CreatedAt
        +time.Time CompletedAt
        +Start()
        +Complete(fileURL string)
        +Fail()
    }

    class MetricType {
        <<enumeration>>
        COUNTER
        GAUGE
        HISTOGRAM
    }

    class TimePeriod {
        <<enumeration>>
        MINUTE
        HOUR
        DAY
        WEEK
        MONTH
    }

    class ReportType {
        <<enumeration>>
        USAGE
        COST
        MODEL
        USER
    }

    class ReportStatus {
        <<enumeration>>
        PENDING
        PROCESSING
        COMPLETED
        FAILED
    }

    Metric --> MetricType
    UsageStats --> TimePeriod
    ModelStats --> TimePeriod
    UserBehavior --> TimePeriod
    CostBreakdown --> TimePeriod
    Report --> ReportType
    Report --> ReportStatus

CREATE TABLE events (
    event_id String,
    event_type String,
    tenant_id String,
    user_id String,
    timestamp DateTime,
    dimensions Map(String, String),
    metrics Map(String, Float64),
    date Date MATERIALIZED toDate(timestamp)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (tenant_id, event_type, timestamp)
TTL timestamp + INTERVAL 30 DAY;

CREATE TABLE metrics_hourly (
    hour DateTime,
    tenant_id String,
    message_count UInt64,
    total_tokens UInt64,
    total_cost Decimal(10, 4),
    active_users UInt64,
    date Date MATERIALIZED toDate(hour)
) ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (tenant_id, hour);

{
    "tenant_id": "tenant_abc",
    "total_conversations": 1000,
    "total_messages": 5000,
    "total_tokens": 1500000,
    "total_cost": 45.50,
    "active_users": 156,
    "period": "day",
    "start_time": "2025-01-20T00:00:00Z",
    "end_time": "2025-01-27T00:00:00Z"
}

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant HTTP as HTTPServer
    participant Usecase as MetricUsecase
    participant Repo as MetricRepository
    participant CH as ClickHouse
    participant MV as metrics_hourly物化视图

    Client->>HTTP: GET /api/v1/stats/usage?tenant_id=xxx&period=day
    Note over HTTP: 参数解析与校验
    HTTP->>HTTP: 解析时间参数（start/end）
    HTTP->>Usecase: GetUsageStats(ctx, tenantID, period, start, end)

    Note over Usecase: 业务逻辑层处理
    Usecase->>Usecase: validateTimePeriod(period)
    alt 周期验证失败
        Usecase-->>HTTP: ErrInvalidTimePeriod
        HTTP-->>Client: 400 Bad Request
    end

    Usecase->>Repo: GetUsageStats(ctx, tenantID, period, start, end)

    Note over Repo: 数据访问层处理
    Repo->>Repo: 构建聚合SQL查询
    Note over Repo: SELECT sum(message_count), sum(total_tokens)<br/>FROM metrics_hourly<br/>WHERE tenant_id = ? AND hour >= ? AND hour < ?

    Repo->>CH: ExecuteQuery(query, params)
    CH->>MV: 查询物化视图表
    Note over MV: 数据已预聚合，直接读取
    MV-->>CH: 返回聚合结果
    CH-->>Repo: 查询结果集

    Repo->>Repo: Scan结果到UsageStats对象
    Repo-->>Usecase: *UsageStats
    Usecase-->>HTTP: *UsageStats
    HTTP->>HTTP: 序列化为JSON
    HTTP-->>Client: 200 OK + JSON响应

// getUsageStats 获取使用统计
func (s *HTTPServer) getUsageStats(c *gin.Context) {
	tenantID := c.Query("tenant_id")
	if tenantID == "" {
		s.respondError(c, http.StatusBadRequest, "tenant_id is required")
		return
	}

	period := c.DefaultQuery("period", "day")
	start, end, err := s.parseTimeRange(c)
	if err != nil {
		s.respondError(c, http.StatusBadRequest, err.Error())
		return
	}

	stats, err := s.service.GetUsageStats(c.Request.Context(), tenantID, period, start, end)
	if err != nil {
		s.handleServiceError(c, err)
		return
	}

	c.JSON(http.StatusOK, stats)
}

// GetUsageStats 获取使用统计
func (uc *MetricUsecase) GetUsageStats(ctx context.Context, tenantID string, period domain.TimePeriod, start, end time.Time) (*domain.UsageStats, error) {
	// 验证时间周期
	if err := uc.validateTimePeriod(period); err != nil {
		return nil, err
	}

	// 获取统计数据
	stats, err := uc.metricRepo.GetUsageStats(ctx, tenantID, period, start, end)
	if err != nil {
		return nil, err
	}

	return stats, nil
}

// validateTimePeriod 验证时间周期
func (uc *MetricUsecase) validateTimePeriod(period domain.TimePeriod) error {
	switch period {
	case domain.PeriodMinute, domain.PeriodHour, domain.PeriodDay, domain.PeriodWeek, domain.PeriodMonth:
		return nil
	default:
		return domain.ErrInvalidTimePeriod
	}
}

// GetUsageStats 获取使用统计
func (r *MetricRepository) GetUsageStats(ctx context.Context, tenantID string, period domain.TimePeriod, start, end time.Time) (*domain.UsageStats, error) {
	// 简化实现：从 ClickHouse 查询聚合数据
	query := `
		SELECT
			COUNT(DISTINCT conversation_id) as total_conversations,
			COUNT(*) as total_messages,
			SUM(tokens_used) as total_tokens,
			SUM(cost_usd) as total_cost,
			COUNT(DISTINCT user_id) as active_users
		FROM message_events
		WHERE tenant_id = ?
			AND created_at >= ?
			AND created_at < ?
	`

	row := r.ch.QueryRow(ctx, query, tenantID, start, end)

	stats := &domain.UsageStats{
		TenantID:  tenantID,
		Period:    period,
		StartTime: start,
		EndTime:   end,
	}

	err := row.Scan(
		&stats.TotalConversations,
		&stats.TotalMessages,
		&stats.TotalTokens,
		&stats.TotalCost,
		&stats.ActiveUsers,
	)

	return stats, err
}

[
    {
        "tenant_id": "tenant_abc",
        "model_name": "gpt-4",
        "provider": "openai",
        "request_count": 5000,
        "total_tokens": 1500000,
        "total_cost": 45.00,
        "avg_latency": 1.5,
        "error_rate": 0.02,
        "period": "day"
    },
    {
        "tenant_id": "tenant_abc",
        "model_name": "gpt-3.5-turbo",
        "provider": "openai",
        "request_count": 20000,
        "total_tokens": 4000000,
        "total_cost": 8.00,
        "avg_latency": 0.8,
        "error_rate": 0.01,
        "period": "day"
    }
]

{
    "tenant_id": "tenant_abc",
    "current_qps": 15.6,
    "current_active_users": 42,
    "current_latency": 1.2,
    "timestamp": "2025-01-27T10:00:00Z"
}

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant HTTP as HTTPServer
    participant Usecase as MetricUsecase
    participant Cache as Redis
    participant Repo as MetricRepository
    participant CH as ClickHouse

    Client->>HTTP: GET /api/v1/stats/realtime?tenant_id=xxx
    HTTP->>Usecase: GetRealtimeStats(ctx, tenantID)

    Note over Usecase: 优先查询缓存
    Usecase->>Cache: GET stats:realtime:tenant_abc

    alt 缓存命中（热路径）
        Cache-->>Usecase: JSON数据（TTL 30s）
        Usecase->>Usecase: 反序列化为RealtimeStats
        Usecase-->>HTTP: *RealtimeStats
        Note over Usecase,HTTP: 延迟 < 10ms
        HTTP-->>Client: 200 OK + JSON响应
    else 缓存未命中（冷路径）
        Cache-->>Usecase: nil（缓存过期或首次查询）

        Usecase->>Repo: GetRealtimeStats(ctx, tenantID)
        Note over Repo: 查询最近1分钟数据
        Repo->>CH: SELECT COUNT(*) / 60.0, uniqExact(user_id)<br/>FROM events WHERE timestamp >= now() - INTERVAL 1 MINUTE
        CH-->>Repo: 聚合结果
        Repo-->>Usecase: *RealtimeStats

        Note over Usecase: 回写缓存
        Usecase->>Usecase: 序列化为JSON
        Usecase->>Cache: SET stats:realtime:tenant_abc, TTL 30s
        Cache-->>Usecase: OK

        Usecase-->>HTTP: *RealtimeStats
        Note over Usecase,HTTP: 延迟 50-100ms
        HTTP-->>Client: 200 OK + JSON响应
    end

// getRealtimeStats 获取实时统计
func (s *HTTPServer) getRealtimeStats(c *gin.Context) {
	tenantID := c.Query("tenant_id")
	if tenantID == "" {
		s.respondError(c, http.StatusBadRequest, "tenant_id is required")
		return
	}

	stats, err := s.service.GetRealtimeStats(c.Request.Context(), tenantID)
	if err != nil {
		s.handleServiceError(c, err)
		return
	}

	c.JSON(http.StatusOK, stats)
}

// GetRealtimeStats 获取实时统计
func (uc *MetricUsecase) GetRealtimeStats(ctx context.Context, tenantID string) (*domain.RealtimeStats, error) {
	stats, err := uc.metricRepo.GetRealtimeStats(ctx, tenantID)
	if err != nil {
		return nil, err
	}

	return stats, nil
}

type CacheClient interface {
    Get(ctx context.Context, key string) (interface{}, error)
    Set(ctx context.Context, key string, value interface{}, ttl time.Duration) error
}

func (uc *MetricUsecase) GetRealtimeStats(ctx context.Context, tenantID string) (*domain.RealtimeStats, error) {
    // 1. 尝试从缓存获取
    cacheKey := fmt.Sprintf("stats:realtime:%s", tenantID)
    if cached, err := uc.cache.Get(ctx, cacheKey); err == nil {
        return cached.(*domain.RealtimeStats), nil
    }

    // 2. 查询数据库
    stats, err := uc.metricRepo.GetRealtimeStats(ctx, tenantID)
    if err != nil {
        return nil, err
    }

    // 3. 写入缓存
    uc.cache.Set(ctx, cacheKey, stats, 30*time.Second)

    return stats, nil
}

// GetRealtimeStats 获取实时统计
func (r *MetricRepository) GetRealtimeStats(ctx context.Context, tenantID string) (*domain.RealtimeStats, error) {
	// 简化实现：查询最近1分钟的数据
	query := `
		SELECT
			COUNT(*) / 60.0 as current_qps,
			COUNT(DISTINCT user_id) as current_active_users,
			AVG(latency_ms) as current_latency
		FROM message_events
		WHERE tenant_id = ?
			AND created_at >= now() - INTERVAL 1 MINUTE
	`

	row := r.ch.QueryRow(ctx, query, tenantID)

	stats := &domain.RealtimeStats{
		TenantID:  tenantID,
		Timestamp: time.Now(),
	}

	err := row.Scan(
		&stats.CurrentQPS,
		&stats.CurrentActiveUsers,
		&stats.CurrentLatency,
	)

	return stats, err
}

{
    "id": "report_123",
    "tenant_id": "tenant_abc",
    "type": "usage",
    "name": "2025年1月使用报表",
    "status": "pending",
    "created_at": "2025-01-27T10:00:00Z"
}

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant HTTP as HTTPServer
    participant Usecase as ReportUsecase
    participant ReportRepo as ReportRepository
    participant MetricRepo as MetricRepository
    participant PG as PostgreSQL
    participant CH as ClickHouse
    participant Goroutine as 异步Goroutine
    participant S3 as 对象存储

    Client->>HTTP: POST /api/v1/reports + JSON Body
    Note over HTTP: 解析并校验请求
    HTTP->>HTTP: 绑定JSON到请求结构体
    HTTP->>Usecase: CreateReport(ctx, tenantID, reportType, name, createdBy)

    Note over Usecase: 同步阶段：创建报表记录
    Usecase->>Usecase: validateReportType(reportType)
    alt 报表类型无效
        Usecase-->>HTTP: ErrInvalidReportType
        HTTP-->>Client: 400 Bad Request
    end

    Usecase->>Usecase: domain.NewReport() 创建报表实体
    Note over Usecase: status=pending, ID=uuid

    Usecase->>ReportRepo: CreateReport(ctx, report)
    ReportRepo->>PG: INSERT INTO reports(id, tenant_id, type, name, status, created_at)
    PG-->>ReportRepo: OK
    ReportRepo-->>Usecase: nil

    Note over Usecase: 启动异步生成
    Usecase->>Goroutine: go generateReportAsync(bg_ctx, report)
    Usecase-->>HTTP: *Report（status=pending）
    HTTP-->>Client: 201 Created + JSON响应

    Note over Goroutine: 异步阶段：生成报表数据
    Goroutine->>Goroutine: 更新状态为processing
    Goroutine->>ReportRepo: UpdateReport(ctx, report)
    ReportRepo->>PG: UPDATE reports SET status='processing'
    PG-->>ReportRepo: OK

    alt 报表类型：usage
        Goroutine->>MetricRepo: GetUsageStats(ctx, tenantID, period, start, end)
        MetricRepo->>CH: 查询聚合数据
        CH-->>MetricRepo: 使用统计数据
        MetricRepo-->>Goroutine: *UsageStats
    else 报表类型：cost
        Goroutine->>MetricRepo: GetCostBreakdown(ctx, tenantID, period, start, end)
        MetricRepo->>CH: 查询成本数据
        CH-->>MetricRepo: 成本分解数据
        MetricRepo-->>Goroutine: *CostBreakdown
    else 其他报表类型
        Note over Goroutine: model/user报表类似流程
    end

    Goroutine->>Goroutine: 序列化为JSON/CSV/Excel
    Goroutine->>S3: 上传文件到对象存储
    S3-->>Goroutine: fileURL

    Goroutine->>Goroutine: report.Complete(fileURL)
    Note over Goroutine: status=completed
    Goroutine->>ReportRepo: UpdateReport(ctx, report)
    ReportRepo->>PG: UPDATE reports SET status='completed', file_url=?, completed_at=?
    PG-->>ReportRepo: OK

    Note over Client: 客户端轮询查询报表状态
    Client->>HTTP: GET /api/v1/reports/{id}
    HTTP->>Usecase: GetReport(ctx, id)
    Usecase->>ReportRepo: GetReport(ctx, id)
    ReportRepo->>PG: SELECT * FROM reports WHERE id=?
    PG-->>ReportRepo: Report记录（status=completed）
    ReportRepo-->>Usecase: *Report
    Usecase-->>HTTP: *Report
    HTTP-->>Client: 200 OK + download_url

// createReport 创建报表
func (s *HTTPServer) createReport(c *gin.Context) {
	var req struct {
		TenantID  string `json:"tenant_id" binding:"required"`
		Type      string `json:"type" binding:"required"`
		Name      string `json:"name" binding:"required"`
		CreatedBy string `json:"created_by" binding:"required"`
	}

	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	report, err := s.service.CreateReport(c.Request.Context(), req.TenantID, req.Type, req.Name, req.CreatedBy)
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
		return
	}

	c.JSON(http.StatusCreated, report)
}

// CreateReport 创建报表
func (uc *ReportUsecase) CreateReport(ctx context.Context, tenantID, reportType, name, createdBy string) (*domain.Report, error) {
	// 验证报表类型
	if err := uc.validateReportType(domain.ReportType(reportType)); err != nil {
		return nil, err
	}

	// 创建报表
	report := domain.NewReport(tenantID, reportType, name, createdBy)

	// 保存到数据库
	if err := uc.reportRepo.CreateReport(ctx, report); err != nil {
		return nil, fmt.Errorf("failed to create report: %w", err)
	}

	// 异步生成报表
	go uc.generateReportAsync(context.Background(), report)

	return report, nil
}

// generateReportAsync 异步生成报表
func (uc *ReportUsecase) generateReportAsync(ctx context.Context, report *domain.Report) {
	// 更新状态为处理中
	report.Status = domain.ReportStatusProcessing
	_ = uc.reportRepo.UpdateReport(ctx, report)

	// 根据报表类型生成数据
	var err error
	switch report.Type {
	case domain.ReportTypeUsage:
		err = uc.generateUsageReport(ctx, report)
	case domain.ReportTypeCost:
		err = uc.generateCostReport(ctx, report)
	case domain.ReportTypeModel:
		err = uc.generateModelReport(ctx, report)
	case domain.ReportTypeUser:
		err = uc.generateUserReport(ctx, report)
	default:
		err = fmt.Errorf("unsupported report type: %s", report.Type)
	}

	// 更新报表状态
	if err != nil {
		report.Fail()
	} else {
		// 生成文件并上传到对象存储（简化实现）
		fileURL := fmt.Sprintf("s3://reports/%s.json", report.ID)
		report.Complete(fileURL)
	}

	_ = uc.reportRepo.UpdateReport(ctx, report)
}

// validateReportType 验证报表类型
func (uc *ReportUsecase) validateReportType(reportType domain.ReportType) error {
	switch reportType {
	case domain.ReportTypeUsage, domain.ReportTypeCost, domain.ReportTypeModel, domain.ReportTypeUser, domain.ReportTypeCustom:
		return nil
	default:
		return domain.ErrInvalidReportType
	}
}

{
    "current_active_users": 125,
    "current_requests_per_sec": 15.6,
    "average_response_time_ms": 450,
    "error_rate": 0.02,
    "today_total_requests": 12000,
    "today_total_tokens": 4500000,
    "today_total_cost": 125.50,
    "today_unique_users": 856,
    "top_models": [
        {"model_name": "gpt-4", "request_count": 5000, "cost": 80.00},
        {"model_name": "gpt-3.5-turbo", "request_count": 7000, "cost": 45.50}
    ],
    "requests_trend": [...],
    "p50_response_time_ms": 350,
    "p95_response_time_ms": 800,
    "p99_response_time_ms": 1500,
    "updated_at": "2025-01-27T10:00:00Z"
}

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant HTTP as HTTPServer
    participant Usecase as DashboardUsecase
    participant Cache as Redis
    participant CH as ClickHouse

    Client->>HTTP: GET /api/v1/dashboard/realtime?tenant_id=xxx
    HTTP->>Usecase: GetDashboardMetrics(ctx, tenantID)

    Note over Usecase: 优先查询缓存
    Usecase->>Cache: GET dashboard:metrics:tenant_abc

    alt 缓存命中
        Cache-->>Usecase: 完整看板数据（TTL 30s）
        Usecase-->>HTTP: *DashboardMetrics
        HTTP-->>Client: 200 OK
    else 缓存未命中
        Cache-->>Usecase: nil

        Note over Usecase: 并发查询9个指标（性能优化）
        par 并发查询1：当前活跃用户
            Usecase->>CH: uniqExact(user_id) WHERE timestamp >= now()-5min
            CH-->>Usecase: 125
        and 并发查询2：今日总请求数
            Usecase->>CH: COUNT(*) WHERE timestamp >= today
            CH-->>Usecase: 12000
        and 并发查询3：今日总Token数
            Usecase->>CH: SUM(tokens) WHERE timestamp >= today
            CH-->>Usecase: 4500000
        and 并发查询4：今日总成本
            Usecase->>CH: SUM(cost) WHERE timestamp >= today
            CH-->>Usecase: 125.50
        and 并发查询5：今日独立用户
            Usecase->>CH: uniqExact(user_id) WHERE timestamp >= today
            CH-->>Usecase: 856
        and 并发查询6：Top模型
            Usecase->>CH: SELECT model, COUNT(*) GROUP BY model LIMIT 5
            CH-->>Usecase: [...]
        and 并发查询7：请求趋势
            Usecase->>CH: SELECT hour, COUNT(*) GROUP BY hour LIMIT 24
            CH-->>Usecase: [...]
        and 并发查询8：平均响应时间
            Usecase->>CH: AVG(latency_ms) WHERE timestamp >= now()-5min
            CH-->>Usecase: 450
        and 并发查询9：错误率
            Usecase->>CH: countIf(error=true)/count() WHERE timestamp >= now()-5min
            CH-->>Usecase: 0.02
        end

        Note over Usecase: 聚合所有结果
        Usecase->>Usecase: 组装DashboardMetrics对象

        Note over Usecase: 写入缓存
        Usecase->>Cache: SET dashboard:metrics:xxx, TTL 30s
        Cache-->>Usecase: OK

        Usecase-->>HTTP: *DashboardMetrics
        HTTP-->>Client: 200 OK
    end

// GetDashboardMetrics 获取实时看板指标
func (uc *RealtimeDashboardUsecase) GetDashboardMetrics(
	ctx context.Context,
	tenantID string,
) (*DashboardMetrics, error) {
	// 1. 尝试从缓存获取
	cacheKey := fmt.Sprintf("dashboard:metrics:%s", tenantID)
	cachedMetrics, err := uc.cacheClient.Get(ctx, cacheKey)
	if err == nil && cachedMetrics != nil {
		metrics := &DashboardMetrics{}
		// 假设缓存存储为JSON
		return metrics, nil
	}

	// 2. 计算实时指标
	now := time.Now()
	today := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())

	// 并发获取各项指标
	type result struct {
		key   string
		value interface{}
		err   error
	}

	resultChan := make(chan result, 10)

	// 获取当前活跃用户
	go func() {
		count, err := uc.getCurrentActiveUsers(ctx, tenantID)
		resultChan <- result{key: "active_users", value: count, err: err}
	}()

	// 获取今日总请求数
	go func() {
		count, err := uc.getTodayTotalRequests(ctx, tenantID, today)
		resultChan <- result{key: "today_requests", value: count, err: err}
	}()

	// 获取今日总Token数
	go func() {
		count, err := uc.getTodayTotalTokens(ctx, tenantID, today)
		resultChan <- result{key: "today_tokens", value: count, err: err}
	}()

	// 获取今日总成本
	go func() {
		cost, err := uc.getTodayTotalCost(ctx, tenantID, today)
		resultChan <- result{key: "today_cost", value: cost, err: err}
	}()

	// 获取今日独立用户数
	go func() {
		count, err := uc.getTodayUniqueUsers(ctx, tenantID, today)
		resultChan <- result{key: "today_users", value: count, err: err}
	}()

	// 获取Top模型
	go func() {
		models, err := uc.getTopModels(ctx, tenantID, today, 5)
		resultChan <- result{key: "top_models", value: models, err: err}
	}()

	// 获取请求趋势
	go func() {
		trend, err := uc.getRequestsTrend(ctx, tenantID, 24)
		resultChan <- result{key: "requests_trend", value: trend, err: err}
	}()

	// 获取平均响应时间
	go func() {
		avgTime, err := uc.getAverageResponseTime(ctx, tenantID, 5*time.Minute)
		resultChan <- result{key: "avg_response_time", value: avgTime, err: err}
	}()

	// 获取错误率
	go func() {
		rate, err := uc.getErrorRate(ctx, tenantID, 5*time.Minute)
		resultChan <- result{key: "error_rate", value: rate, err: err}
	}()

	// 收集结果
	metrics := &DashboardMetrics{
		UpdatedAt: now,
	}

	for i := 0; i < 9; i++ {
		res := <-resultChan
		if res.err != nil {
			continue
		}

		switch res.key {
		case "active_users":
			metrics.CurrentActiveUsers = res.value.(int64)
		case "today_requests":
			metrics.TodayTotalRequests = res.value.(int64)
		case "today_tokens":
			metrics.TodayTotalTokens = res.value.(int64)
		case "today_cost":
			metrics.TodayTotalCost = res.value.(float64)
		case "today_users":
			metrics.TodayUniqueUsers = res.value.(int64)
		case "top_models":
			metrics.TopModels = res.value.([]ModelUsage)
		case "requests_trend":
			metrics.RequestsTrend = res.value.([]TimeSeriesPoint)
		case "avg_response_time":
			metrics.AverageResponseTime = res.value.(float64)
		case "error_rate":
			metrics.ErrorRate = res.value.(float64)
		}
	}

	// 3. 缓存结果（30秒）
	uc.cacheClient.Set(ctx, cacheKey, metrics, 30*time.Second)

	return metrics, nil
}

// getCurrentActiveUsers 获取当前活跃用户数
func (uc *RealtimeDashboardUsecase) getCurrentActiveUsers(
	ctx context.Context,
	tenantID string,
) (int64, error) {
	query := `
		SELECT uniqExact(user_id) as count
		FROM usage_metrics
		WHERE tenant_id = ?
		  AND timestamp >= now() - INTERVAL 5 MINUTE
	`

	var count int64
	err := uc.clickhouseClient.QueryRow(ctx, query, tenantID).Scan(&count)
	return count, err
}

flowchart TB
    subgraph DataFlow["数据流转"]
        Kafka[Kafka事件流] -->|订阅消费| Consumer[事件消费者]
        Consumer -->|批量写入| Events[events原始表]
        Events -->|触发| MV[物化视图]
        MV -->|增量更新| MetricsHourly[metrics_hourly表]
        MV -->|增量更新| MetricsDaily[metrics_daily表]

        Kafka -->|实时订阅| RealTimeAgg[实时聚合器]
        RealTimeAgg -->|内存计算| MemBuffer[内存缓冲区]
        MemBuffer -->|5秒flush| RedisCache[Redis缓存]
    end

    subgraph QueryPath["查询路径"]
        API[API请求] -->|查询| CacheLayer{缓存层}
        CacheLayer -->|命中| RedisCache
        CacheLayer -->|未命中| AggTables[聚合表]
        AggTables -->|快速查询| MetricsHourly
        AggTables -->|历史查询| MetricsDaily

        RedisCache -->|返回| Response[API响应]
        MetricsHourly -->|返回| Response
        MetricsDaily -->|返回| Response
    end

    style Kafka fill:#e3f2fd
    style Events fill:#fff3e0
    style MV fill:#e8f5e9
    style MetricsHourly fill:#e0f2f1
    style MetricsDaily fill:#f3e5f5
    style RedisCache fill:#ffebee