package handler import ( "net/http" "strconv" "time" "github.com/gin-gonic/gin" "gorm.io/gorm" "silk-server-go/internal/middleware" "silk-server-go/internal/model" "silk-server-go/internal/service" ) // RegisterTelemetryRoutes 注册遥测数据查询路由 func RegisterTelemetryRoutes(rg *gin.RouterGroup, db *gorm.DB, iotdb *service.IoTDBService) { readPerm := middleware.RequirePermission(db, "device:read") rg.GET("/telemetry", readPerm, listTelemetry(db, iotdb)) rg.GET("/telemetry/:deviceKey/metrics", readPerm, listMetrics(iotdb, db)) rg.GET("/telemetry/:deviceKey/latest", readPerm, latestTelemetry(iotdb, db)) rg.GET("/telemetry/:deviceKey/:metric/history", readPerm, historyBucket(iotdb, db)) } // GET /telemetry?deviceKey=&metric=&from=&to=&limit= func listTelemetry(db *gorm.DB, iotdb *service.IoTDBService) gin.HandlerFunc { return func(c *gin.Context) { deviceKey := c.Query("deviceKey") metric := c.Query("metric") from := c.Query("from") to := c.Query("to") limit, _ := strconv.Atoi(c.DefaultQuery("limit", "2000")) if limit <= 0 { limit = 2000 } // 优先 IoTDB if iotdb.IsAvailable() && deviceKey != "" && metric != "" { fromTime, toTime := parseTimeRange(from, to) rows, err := iotdb.QueryHistory(deviceKey, metric, fromTime, toTime, limit) if err == nil { c.JSON(http.StatusOK, rows) return } } // 降级 PostgreSQL q := db.Model(&model.Telemetry{}).Order("timestamp DESC").Limit(limit) if deviceKey != "" { q = q.Where("device_key = ?", deviceKey) } if metric != "" { q = q.Where("metric = ?", metric) } if from != "" || to != "" { fromVal, toVal := parseTimeRangeStr(from, to) q = q.Where("timestamp BETWEEN ? AND ?", fromVal, toVal) } var rows []model.Telemetry q.Find(&rows) c.JSON(http.StatusOK, rows) } } // GET /telemetry/:deviceKey/metrics func listMetrics(iotdb *service.IoTDBService, db *gorm.DB) gin.HandlerFunc { return func(c *gin.Context) { deviceKey := c.Param("deviceKey") // 优先 IoTDB if iotdb.IsAvailable() { metrics, err := iotdb.ListMetrics(deviceKey) if err == nil && len(metrics) > 0 { c.JSON(http.StatusOK, metrics) return } } // 降级 PG var metrics []string db.Model(&model.Telemetry{}).Distinct("metric").Where("device_key = ?", deviceKey).Pluck("metric", &metrics) c.JSON(http.StatusOK, metrics) } } // GET /telemetry/:deviceKey/latest func latestTelemetry(iotdb *service.IoTDBService, db *gorm.DB) gin.HandlerFunc { return func(c *gin.Context) { deviceKey := c.Param("deviceKey") // 查询 PG 中该设备每个指标的最新记录 var records []model.Telemetry if err := db.Raw(`SELECT DISTINCT ON (metric) * FROM telemetry WHERE device_key = ? ORDER BY metric, timestamp DESC`, deviceKey).Scan(&records).Error; err != nil || len(records) == 0 { c.JSON(http.StatusNotFound, gin.H{"error": "未找到遥测数据"}) return } c.JSON(http.StatusOK, records) } } // GET /telemetry/:deviceKey/:metric/history?from=&to=&bucketMin= func historyBucket(iotdb *service.IoTDBService, db *gorm.DB) gin.HandlerFunc { return func(c *gin.Context) { deviceKey := c.Param("deviceKey") metric := c.Param("metric") from := c.Query("from") to := c.Query("to") if from == "" || to == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "from/to required"}) return } bucketMin, _ := strconv.Atoi(c.DefaultQuery("bucketMin", "5")) if bucketMin <= 0 { bucketMin = 5 } fromTime, _ := time.Parse(time.RFC3339, from) toTime, _ := time.Parse(time.RFC3339, to) // 优先 IoTDB if iotdb.IsAvailable() { rows, err := iotdb.AggregateByBucket(deviceKey, metric, fromTime, toTime, bucketMin) if err == nil && len(rows) > 0 { c.JSON(http.StatusOK, rows) return } } // 降级 PG type BucketResult struct { Bucket time.Time `json:"time"` Avg float64 `json:"value"` } var results []BucketResult sql := `SELECT time_bucket(?, timestamp) AS bucket, AVG(value) AS avg FROM telemetry WHERE device_key = ? AND metric = ? AND timestamp BETWEEN ? AND ? GROUP BY 1 ORDER BY 1 ASC` if err := db.Raw(sql, strconv.Itoa(bucketMin)+" minutes", deviceKey, metric, fromTime, toTime).Scan(&results).Error; err != nil || len(results) == 0 { // time_bucket 不可用(无 TimescaleDB),降级为原始数据 var raw []model.Telemetry db.Where("device_key = ? AND metric = ? AND timestamp BETWEEN ? AND ?", deviceKey, metric, fromTime, toTime). Order("timestamp ASC").Limit(500).Find(&raw) for _, r := range raw { results = append(results, BucketResult{Bucket: r.Timestamp, Avg: r.Value}) } } c.JSON(http.StatusOK, results) } } // --- 辅助函数 --- func parseTimeRange(from, to string) (time.Time, time.Time) { var fromTime, toTime time.Time if from != "" { fromTime, _ = time.Parse(time.RFC3339, from) } if to != "" { toTime, _ = time.Parse(time.RFC3339, to) } else { toTime = time.Now() } return fromTime, toTime } func parseTimeRangeStr(from, to string) (string, string) { if from == "" { from = "1970-01-01T00:00:00Z" } if to == "" { to = time.Now().Format(time.RFC3339) } return from, to }