silk/server-go/internal/handler/telemetry.go

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
		}
		if deviceKey != "" && !requireDeviceKeyAccess(c, db, deviceKey) {
			return
		}

		// 优先 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{}).
			Joins("JOIN devices d ON d.device_key = telemetry.device_key").
			Order("telemetry.timestamp DESC").Limit(limit)
		q = applyRoomScope(q, c, "d.room_id")
		if deviceKey != "" {
			q = q.Where("telemetry.device_key = ?", deviceKey)
		}
		if metric != "" {
			q = q.Where("telemetry.metric = ?", metric)
		}
		if from != "" || to != "" {
			fromVal, toVal := parseTimeRangeStr(from, to)
			q = q.Where("telemetry.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")
		if !requireDeviceKeyAccess(c, db, deviceKey) {
			return
		}

		// 优先 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")
		if !requireDeviceKeyAccess(c, db, deviceKey) {
			return
		}

		// 查询 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 !requireDeviceKeyAccess(c, db, deviceKey) {
			return
		}
		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
}