From 955af6b3ec58ff799258118f9a01e80c1a0c442c Mon Sep 17 00:00:00 2001
From: yangjianbo <yangjianbo@leagsoft.com>
Date: Mon, 12 Jan 2026 10:59:52 +0800
Subject: [PATCH] =?UTF-8?q?fix(=E4=BB=AA=E8=A1=A8=E7=9B=98):=20=E6=B7=BB?=
 =?UTF-8?q?=E5=8A=A0=E8=81=9A=E5=90=88=E5=92=8C=E5=9B=9E=E5=A1=AB=E6=93=8D?=
 =?UTF-8?q?=E4=BD=9C=E7=9A=84=E6=89=A7=E8=A1=8C=E6=97=B6=E9=97=B4=E6=97=A5?=
 =?UTF-8?q?=E5=BF=97?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../service/dashboard_aggregation_service.go  | 24 +++++++++++++++----
 1 file changed, 20 insertions(+), 4 deletions(-)

diff --git a/backend/internal/service/dashboard_aggregation_service.go b/backend/internal/service/dashboard_aggregation_service.go
index 0d1cec57..da5c0e7d 100644
--- a/backend/internal/service/dashboard_aggregation_service.go
+++ b/backend/internal/service/dashboard_aggregation_service.go
@@ -134,6 +134,7 @@ func (s *DashboardAggregationService) runScheduledAggregation() {
 	}
 	defer atomic.StoreInt32(&s.running, 0)
 
+	jobStart := time.Now().UTC()
 	ctx, cancel := context.WithTimeout(context.Background(), defaultDashboardAggregationTimeout)
 	defer cancel()
 
@@ -162,9 +163,16 @@ func (s *DashboardAggregationService) runScheduledAggregation() {
 		return
 	}
 
-	if err := s.repo.UpdateAggregationWatermark(ctx, now); err != nil {
-		log.Printf("[DashboardAggregation] 更新水位失败: %v", err)
+	updateErr := s.repo.UpdateAggregationWatermark(ctx, now)
+	if updateErr != nil {
+		log.Printf("[DashboardAggregation] 更新水位失败: %v", updateErr)
 	}
+	log.Printf("[DashboardAggregation] 聚合完成 (start=%s end=%s duration=%s watermark_updated=%t)",
+		start.Format(time.RFC3339),
+		now.Format(time.RFC3339),
+		time.Since(jobStart).String(),
+		updateErr == nil,
+	)
 
 	s.maybeCleanupRetention(ctx, now)
 }
@@ -175,6 +183,7 @@ func (s *DashboardAggregationService) backfillRange(ctx context.Context, start,
 	}
 	defer atomic.StoreInt32(&s.running, 0)
 
+	jobStart := time.Now().UTC()
 	startUTC := start.UTC()
 	endUTC := end.UTC()
 	if !endUTC.After(startUTC) {
@@ -193,9 +202,16 @@ func (s *DashboardAggregationService) backfillRange(ctx context.Context, start,
 		cursor = windowEnd
 	}
 
-	if err := s.repo.UpdateAggregationWatermark(ctx, endUTC); err != nil {
-		log.Printf("[DashboardAggregation] 更新水位失败: %v", err)
+	updateErr := s.repo.UpdateAggregationWatermark(ctx, endUTC)
+	if updateErr != nil {
+		log.Printf("[DashboardAggregation] 更新水位失败: %v", updateErr)
 	}
+	log.Printf("[DashboardAggregation] 回填聚合完成 (start=%s end=%s duration=%s watermark_updated=%t)",
+		startUTC.Format(time.RFC3339),
+		endUTC.Format(time.RFC3339),
+		time.Since(jobStart).String(),
+		updateErr == nil,
+	)
 
 	s.maybeCleanupRetention(ctx, endUTC)
 	return nil