package events import ( "context" "encoding/json" "fmt" "time" "go.uber.org/zap" "tyapi-server/internal/domains/statistics/events" "tyapi-server/internal/domains/statistics/repositories" "tyapi-server/internal/infrastructure/statistics/cache" ) // StatisticsEventHandler 统计事件处理器 type StatisticsEventHandler struct { metricRepo repositories.StatisticsRepository reportRepo repositories.StatisticsReportRepository dashboardRepo repositories.StatisticsDashboardRepository cache *cache.RedisStatisticsCache logger *zap.Logger } // NewStatisticsEventHandler 创建统计事件处理器 func NewStatisticsEventHandler( metricRepo repositories.StatisticsRepository, reportRepo repositories.StatisticsReportRepository, dashboardRepo repositories.StatisticsDashboardRepository, cache *cache.RedisStatisticsCache, logger *zap.Logger, ) *StatisticsEventHandler { return &StatisticsEventHandler{ metricRepo: metricRepo, reportRepo: reportRepo, dashboardRepo: dashboardRepo, cache: cache, logger: logger, } } // HandleMetricCreatedEvent 处理指标创建事件 func (h *StatisticsEventHandler) HandleMetricCreatedEvent(ctx context.Context, event *events.MetricCreatedEvent) error { h.logger.Info("处理指标创建事件", zap.String("metric_id", event.MetricID), zap.String("metric_type", event.MetricType), zap.String("metric_name", event.MetricName), zap.Float64("value", event.Value)) // 更新实时指标缓存 err := h.cache.UpdateRealtimeMetric(ctx, event.MetricType, event.MetricName, event.Value, 1*time.Hour) if err != nil { h.logger.Error("更新实时指标缓存失败", zap.Error(err)) // 不返回错误,避免影响主流程 } // 清理相关缓存 err = h.cache.DeleteMetricsByType(ctx, event.MetricType) if err != nil { h.logger.Error("清理指标类型缓存失败", zap.Error(err)) } return nil } // HandleMetricUpdatedEvent 处理指标更新事件 func (h *StatisticsEventHandler) HandleMetricUpdatedEvent(ctx context.Context, event *events.MetricUpdatedEvent) error { h.logger.Info("处理指标更新事件", zap.String("metric_id", event.MetricID), zap.Float64("old_value", event.OldValue), zap.Float64("new_value", event.NewValue)) // 获取指标信息 metric, err := h.metricRepo.FindByID(ctx, event.MetricID) if err != nil { h.logger.Error("查询指标失败", zap.Error(err)) return err } // 更新实时指标缓存 err = h.cache.UpdateRealtimeMetric(ctx, metric.MetricType, metric.MetricName, event.NewValue, 1*time.Hour) if err != nil { h.logger.Error("更新实时指标缓存失败", zap.Error(err)) } // 清理相关缓存 err = h.cache.DeleteMetric(ctx, event.MetricID) if err != nil { h.logger.Error("清理指标缓存失败", zap.Error(err)) } err = h.cache.DeleteMetricsByType(ctx, metric.MetricType) if err != nil { h.logger.Error("清理指标类型缓存失败", zap.Error(err)) } return nil } // HandleMetricAggregatedEvent 处理指标聚合事件 func (h *StatisticsEventHandler) HandleMetricAggregatedEvent(ctx context.Context, event *events.MetricAggregatedEvent) error { h.logger.Info("处理指标聚合事件", zap.String("metric_type", event.MetricType), zap.String("dimension", event.Dimension), zap.Int("record_count", event.RecordCount), zap.Float64("total_value", event.TotalValue)) // 清理相关缓存 err := h.cache.ClearRealtimeCache(ctx) if err != nil { h.logger.Error("清理实时缓存失败", zap.Error(err)) } err = h.cache.ClearMetricsCache(ctx) if err != nil { h.logger.Error("清理指标缓存失败", zap.Error(err)) } return nil } // HandleReportCreatedEvent 处理报告创建事件 func (h *StatisticsEventHandler) HandleReportCreatedEvent(ctx context.Context, event *events.ReportCreatedEvent) error { h.logger.Info("处理报告创建事件", zap.String("report_id", event.ReportID), zap.String("report_type", event.ReportType), zap.String("title", event.Title)) // 获取报告信息 report, err := h.reportRepo.FindByID(ctx, event.ReportID) if err != nil { h.logger.Error("查询报告失败", zap.Error(err)) return err } // 设置报告缓存 err = h.cache.SetReport(ctx, report, 24*time.Hour) if err != nil { h.logger.Error("设置报告缓存失败", zap.Error(err)) } return nil } // HandleReportGenerationStartedEvent 处理报告生成开始事件 func (h *StatisticsEventHandler) HandleReportGenerationStartedEvent(ctx context.Context, event *events.ReportGenerationStartedEvent) error { h.logger.Info("处理报告生成开始事件", zap.String("report_id", event.ReportID), zap.String("generated_by", event.GeneratedBy)) // 获取报告信息 report, err := h.reportRepo.FindByID(ctx, event.ReportID) if err != nil { h.logger.Error("查询报告失败", zap.Error(err)) return err } // 更新报告缓存 err = h.cache.SetReport(ctx, report, 24*time.Hour) if err != nil { h.logger.Error("更新报告缓存失败", zap.Error(err)) } return nil } // HandleReportCompletedEvent 处理报告完成事件 func (h *StatisticsEventHandler) HandleReportCompletedEvent(ctx context.Context, event *events.ReportCompletedEvent) error { h.logger.Info("处理报告完成事件", zap.String("report_id", event.ReportID), zap.Int("content_size", event.ContentSize)) // 获取报告信息 report, err := h.reportRepo.FindByID(ctx, event.ReportID) if err != nil { h.logger.Error("查询报告失败", zap.Error(err)) return err } // 更新报告缓存 err = h.cache.SetReport(ctx, report, 7*24*time.Hour) // 报告完成后缓存7天 if err != nil { h.logger.Error("更新报告缓存失败", zap.Error(err)) } return nil } // HandleReportFailedEvent 处理报告失败事件 func (h *StatisticsEventHandler) HandleReportFailedEvent(ctx context.Context, event *events.ReportFailedEvent) error { h.logger.Info("处理报告失败事件", zap.String("report_id", event.ReportID), zap.String("reason", event.Reason)) // 获取报告信息 report, err := h.reportRepo.FindByID(ctx, event.ReportID) if err != nil { h.logger.Error("查询报告失败", zap.Error(err)) return err } // 更新报告缓存 err = h.cache.SetReport(ctx, report, 1*time.Hour) // 失败报告只缓存1小时 if err != nil { h.logger.Error("更新报告缓存失败", zap.Error(err)) } return nil } // HandleDashboardCreatedEvent 处理仪表板创建事件 func (h *StatisticsEventHandler) HandleDashboardCreatedEvent(ctx context.Context, event *events.DashboardCreatedEvent) error { h.logger.Info("处理仪表板创建事件", zap.String("dashboard_id", event.DashboardID), zap.String("name", event.Name), zap.String("user_role", event.UserRole)) // 获取仪表板信息 dashboard, err := h.dashboardRepo.FindByID(ctx, event.DashboardID) if err != nil { h.logger.Error("查询仪表板失败", zap.Error(err)) return err } // 设置仪表板缓存 err = h.cache.SetDashboard(ctx, dashboard, 24*time.Hour) if err != nil { h.logger.Error("设置仪表板缓存失败", zap.Error(err)) } // 清理仪表板数据缓存 err = h.cache.DeleteDashboardData(ctx, event.UserRole) if err != nil { h.logger.Error("清理仪表板数据缓存失败", zap.Error(err)) } return nil } // HandleDashboardUpdatedEvent 处理仪表板更新事件 func (h *StatisticsEventHandler) HandleDashboardUpdatedEvent(ctx context.Context, event *events.DashboardUpdatedEvent) error { h.logger.Info("处理仪表板更新事件", zap.String("dashboard_id", event.DashboardID), zap.String("updated_by", event.UpdatedBy)) // 获取仪表板信息 dashboard, err := h.dashboardRepo.FindByID(ctx, event.DashboardID) if err != nil { h.logger.Error("查询仪表板失败", zap.Error(err)) return err } // 更新仪表板缓存 err = h.cache.SetDashboard(ctx, dashboard, 24*time.Hour) if err != nil { h.logger.Error("更新仪表板缓存失败", zap.Error(err)) } // 清理仪表板数据缓存 err = h.cache.DeleteDashboardData(ctx, dashboard.UserRole) if err != nil { h.logger.Error("清理仪表板数据缓存失败", zap.Error(err)) } return nil } // HandleDashboardActivatedEvent 处理仪表板激活事件 func (h *StatisticsEventHandler) HandleDashboardActivatedEvent(ctx context.Context, event *events.DashboardActivatedEvent) error { h.logger.Info("处理仪表板激活事件", zap.String("dashboard_id", event.DashboardID), zap.String("activated_by", event.ActivatedBy)) // 获取仪表板信息 dashboard, err := h.dashboardRepo.FindByID(ctx, event.DashboardID) if err != nil { h.logger.Error("查询仪表板失败", zap.Error(err)) return err } // 更新仪表板缓存 err = h.cache.SetDashboard(ctx, dashboard, 24*time.Hour) if err != nil { h.logger.Error("更新仪表板缓存失败", zap.Error(err)) } // 清理仪表板数据缓存 err = h.cache.DeleteDashboardData(ctx, dashboard.UserRole) if err != nil { h.logger.Error("清理仪表板数据缓存失败", zap.Error(err)) } return nil } // HandleDashboardDeactivatedEvent 处理仪表板停用事件 func (h *StatisticsEventHandler) HandleDashboardDeactivatedEvent(ctx context.Context, event *events.DashboardDeactivatedEvent) error { h.logger.Info("处理仪表板停用事件", zap.String("dashboard_id", event.DashboardID), zap.String("deactivated_by", event.DeactivatedBy)) // 获取仪表板信息 dashboard, err := h.dashboardRepo.FindByID(ctx, event.DashboardID) if err != nil { h.logger.Error("查询仪表板失败", zap.Error(err)) return err } // 更新仪表板缓存 err = h.cache.SetDashboard(ctx, dashboard, 24*time.Hour) if err != nil { h.logger.Error("更新仪表板缓存失败", zap.Error(err)) } // 清理仪表板数据缓存 err = h.cache.DeleteDashboardData(ctx, dashboard.UserRole) if err != nil { h.logger.Error("清理仪表板数据缓存失败", zap.Error(err)) } return nil } // ================ 事件分发器 ================ // EventDispatcher 事件分发器 type EventDispatcher struct { handlers map[string][]func(context.Context, interface{}) error logger *zap.Logger } // NewEventDispatcher 创建事件分发器 func NewEventDispatcher(logger *zap.Logger) *EventDispatcher { return &EventDispatcher{ handlers: make(map[string][]func(context.Context, interface{}) error), logger: logger, } } // RegisterHandler 注册事件处理器 func (d *EventDispatcher) RegisterHandler(eventType string, handler func(context.Context, interface{}) error) { if d.handlers[eventType] == nil { d.handlers[eventType] = make([]func(context.Context, interface{}) error, 0) } d.handlers[eventType] = append(d.handlers[eventType], handler) } // Dispatch 分发事件 func (d *EventDispatcher) Dispatch(ctx context.Context, event interface{}) error { // 获取事件类型 eventType := d.getEventType(event) if eventType == "" { return fmt.Errorf("无法确定事件类型") } // 获取处理器 handlers := d.handlers[eventType] if len(handlers) == 0 { d.logger.Warn("没有找到事件处理器", zap.String("event_type", eventType)) return nil } // 执行所有处理器 for _, handler := range handlers { err := handler(ctx, event) if err != nil { d.logger.Error("事件处理器执行失败", zap.String("event_type", eventType), zap.Error(err)) // 继续执行其他处理器 } } return nil } // getEventType 获取事件类型 func (d *EventDispatcher) getEventType(event interface{}) string { switch event.(type) { case *events.MetricCreatedEvent: return string(events.MetricCreatedEventType) case *events.MetricUpdatedEvent: return string(events.MetricUpdatedEventType) case *events.MetricAggregatedEvent: return string(events.MetricAggregatedEventType) case *events.ReportCreatedEvent: return string(events.ReportCreatedEventType) case *events.ReportGenerationStartedEvent: return string(events.ReportGenerationStartedEventType) case *events.ReportCompletedEvent: return string(events.ReportCompletedEventType) case *events.ReportFailedEvent: return string(events.ReportFailedEventType) case *events.DashboardCreatedEvent: return string(events.DashboardCreatedEventType) case *events.DashboardUpdatedEvent: return string(events.DashboardUpdatedEventType) case *events.DashboardActivatedEvent: return string(events.DashboardActivatedEventType) case *events.DashboardDeactivatedEvent: return string(events.DashboardDeactivatedEventType) default: return "" } } // ================ 事件监听器 ================ // EventListener 事件监听器 type EventListener struct { dispatcher *EventDispatcher logger *zap.Logger } // NewEventListener 创建事件监听器 func NewEventListener(dispatcher *EventDispatcher, logger *zap.Logger) *EventListener { return &EventListener{ dispatcher: dispatcher, logger: logger, } } // Listen 监听事件 func (l *EventListener) Listen(ctx context.Context, eventData []byte) error { // 解析事件数据 var baseEvent events.BaseStatisticsEvent err := json.Unmarshal(eventData, &baseEvent) if err != nil { return fmt.Errorf("解析事件数据失败: %w", err) } // 根据事件类型创建具体事件 event, err := l.createEventByType(baseEvent.Type, eventData) if err != nil { return fmt.Errorf("创建事件失败: %w", err) } // 分发事件 err = l.dispatcher.Dispatch(ctx, event) if err != nil { return fmt.Errorf("分发事件失败: %w", err) } return nil } // createEventByType 根据事件类型创建具体事件 func (l *EventListener) createEventByType(eventType string, eventData []byte) (interface{}, error) { switch eventType { case string(events.MetricCreatedEventType): var event events.MetricCreatedEvent err := json.Unmarshal(eventData, &event) return &event, err case string(events.MetricUpdatedEventType): var event events.MetricUpdatedEvent err := json.Unmarshal(eventData, &event) return &event, err case string(events.MetricAggregatedEventType): var event events.MetricAggregatedEvent err := json.Unmarshal(eventData, &event) return &event, err case string(events.ReportCreatedEventType): var event events.ReportCreatedEvent err := json.Unmarshal(eventData, &event) return &event, err case string(events.ReportGenerationStartedEventType): var event events.ReportGenerationStartedEvent err := json.Unmarshal(eventData, &event) return &event, err case string(events.ReportCompletedEventType): var event events.ReportCompletedEvent err := json.Unmarshal(eventData, &event) return &event, err case string(events.ReportFailedEventType): var event events.ReportFailedEvent err := json.Unmarshal(eventData, &event) return &event, err case string(events.DashboardCreatedEventType): var event events.DashboardCreatedEvent err := json.Unmarshal(eventData, &event) return &event, err case string(events.DashboardUpdatedEventType): var event events.DashboardUpdatedEvent err := json.Unmarshal(eventData, &event) return &event, err case string(events.DashboardActivatedEventType): var event events.DashboardActivatedEvent err := json.Unmarshal(eventData, &event) return &event, err case string(events.DashboardDeactivatedEventType): var event events.DashboardDeactivatedEvent err := json.Unmarshal(eventData, &event) return &event, err default: return nil, fmt.Errorf("未知的事件类型: %s", eventType) } }