From 1a53011408936706b92079b380acebc62f0e45f8 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 23:30:54 +0800 Subject: [PATCH] feat: expose daily metric query api --- deploy/portainer/docker-compose-go.yml | 1 + go/vehicle-gateway/cmd/realtime-api/main.go | 32 ++- go/vehicle-gateway/internal/stats/query.go | 241 ++++++++++++++++++ .../internal/stats/query_test.go | 100 ++++++++ 4 files changed, 373 insertions(+), 1 deletion(-) create mode 100644 go/vehicle-gateway/internal/stats/query.go create mode 100644 go/vehicle-gateway/internal/stats/query_test.go diff --git a/deploy/portainer/docker-compose-go.yml b/deploy/portainer/docker-compose-go.yml index 88dc3aa5..4bbd5ec8 100644 --- a/deploy/portainer/docker-compose-go.yml +++ b/deploy/portainer/docker-compose-go.yml @@ -95,6 +95,7 @@ services: REDIS_PASSWORD: ${REDIS_PASSWORD:?set REDIS_PASSWORD} REDIS_DB: ${REDIS_DB:-50} ONLINE_TTL_SECONDS: ${ONLINE_TTL_SECONDS:-600} + MYSQL_DSN: ${MYSQL_DSN:-} ports: - "${GO_REALTIME_HTTP_PORT:-20210}:20210" diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index 637abce9..8d2dba67 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -2,6 +2,7 @@ package main import ( "context" + "database/sql" "encoding/json" "net/http" "os" @@ -11,12 +12,14 @@ import ( "syscall" "time" + _ "github.com/go-sql-driver/mysql" "github.com/redis/go-redis/v9" "github.com/segmentio/kafka-go" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats" ) func main() { @@ -45,9 +48,36 @@ func main() { logger.Warn("KAFKA_BROKERS is empty; realtime api will serve existing redis data only") } + mux := http.NewServeMux() + mux.Handle("/api/realtime/vehicles/", realtime.NewHandler(repository)) + closeStats := func() {} + if dsn := strings.TrimSpace(os.Getenv("MYSQL_DSN")); dsn != "" { + db, err := sql.Open("mysql", dsn) + if err != nil { + logger.Error("stats mysql open failed", "error", err) + os.Exit(1) + } + if err := db.PingContext(ctx); err != nil { + _ = db.Close() + logger.Error("stats mysql ping failed", "error", err) + os.Exit(1) + } + closeStats = func() { _ = db.Close() } + mux.Handle("/api/stats/daily-metrics", stats.NewMetricHandler(stats.NewMetricRepository(db))) + logger.Info("stats mysql query enabled") + } else { + mux.HandleFunc("/api/stats/daily-metrics", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + _ = json.NewEncoder(w).Encode(map[string]any{"error": "MYSQL_DSN is not configured"}) + }) + logger.Warn("MYSQL_DSN is empty; stats query api disabled") + } + defer closeStats() + server := &http.Server{ Addr: env("HTTP_ADDR", ":20210"), - Handler: realtime.NewHandler(repository), + Handler: mux, ReadHeaderTimeout: 5 * time.Second, } go func() { diff --git a/go/vehicle-gateway/internal/stats/query.go b/go/vehicle-gateway/internal/stats/query.go new file mode 100644 index 00000000..51f4c9ca --- /dev/null +++ b/go/vehicle-gateway/internal/stats/query.go @@ -0,0 +1,241 @@ +package stats + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + "strings" +) + +type Queryer interface { + QueryContext(context.Context, string, ...any) (*sql.Rows, error) +} + +type MetricQuery struct { + VIN string + Protocol string + MetricKey string + DateFrom string + DateTo string + Limit int + Offset int +} + +type MetricRow struct { + VIN string `json:"vin"` + StatDate string `json:"stat_date"` + Protocol string `json:"protocol"` + MetricKey string `json:"metric_key"` + MetricValue float64 `json:"metric_value"` + MetricUnit string `json:"metric_unit"` + FirstTotalMileageKM *float64 `json:"first_total_mileage_km,omitempty"` + LatestTotalMileageKM *float64 `json:"latest_total_mileage_km,omitempty"` + SampleCount int64 `json:"sample_count"` + CalculationMethod string `json:"calculation_method"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} + +type MetricRepository struct { + db Queryer +} + +func NewMetricRepository(db Queryer) *MetricRepository { + if db == nil { + panic("metric query db must not be nil") + } + return &MetricRepository{db: db} +} + +func (r *MetricRepository) Query(ctx context.Context, query MetricQuery) ([]MetricRow, error) { + query = normalizeMetricQuery(query) + sqlText, args := buildMetricSQL(query) + rows, err := r.db.QueryContext(ctx, sqlText, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []MetricRow + for rows.Next() { + var row MetricRow + var first sql.NullFloat64 + var latest sql.NullFloat64 + if err := rows.Scan( + &row.VIN, + &row.StatDate, + &row.Protocol, + &row.MetricKey, + &row.MetricValue, + &row.MetricUnit, + &first, + &latest, + &row.SampleCount, + &row.CalculationMethod, + &row.CreatedAt, + &row.UpdatedAt, + ); err != nil { + return nil, err + } + if first.Valid { + row.FirstTotalMileageKM = &first.Float64 + } + if latest.Valid { + row.LatestTotalMileageKM = &latest.Float64 + } + out = append(out, row) + } + return out, rows.Err() +} + +func normalizeMetricQuery(query MetricQuery) MetricQuery { + query.VIN = strings.TrimSpace(query.VIN) + query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol)) + query.MetricKey = strings.TrimSpace(query.MetricKey) + query.DateFrom = strings.TrimSpace(query.DateFrom) + query.DateTo = strings.TrimSpace(query.DateTo) + if query.Limit <= 0 { + query.Limit = 50 + } + return query +} + +func buildMetricSQL(query MetricQuery) (string, []any) { + var where []string + var args []any + add := func(clause string, value any) { + where = append(where, clause) + args = append(args, value) + } + if query.VIN != "" { + add("vin = ?", query.VIN) + } + if query.Protocol != "" { + add("protocol = ?", query.Protocol) + } + if query.MetricKey != "" { + add("metric_key = ?", query.MetricKey) + } + if query.DateFrom != "" { + add("stat_date >= ?", query.DateFrom) + } + if query.DateTo != "" { + add("stat_date <= ?", query.DateTo) + } + sqlText := `SELECT vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric` + if len(where) > 0 { + sqlText += " WHERE " + strings.Join(where, " AND ") + } + sqlText += " ORDER BY stat_date DESC, vin ASC, protocol ASC, metric_key ASC LIMIT ? OFFSET ?" + args = append(args, query.Limit, query.Offset) + return sqlText, args +} + +type MetricHandler struct { + repository *MetricRepository +} + +func NewMetricHandler(repository *MetricRepository) *MetricHandler { + if repository == nil { + panic("metric repository must not be nil") + } + return &MetricHandler{repository: repository} +} + +func (h *MetricHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeMetricError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if strings.Trim(r.URL.Path, "/") != "api/stats/daily-metrics" { + writeMetricError(w, http.StatusNotFound, "route not found") + return + } + query, err := parseMetricQuery(r) + if err != nil { + writeMetricError(w, http.StatusBadRequest, err.Error()) + return + } + rows, err := h.repository.Query(r.Context(), query) + if err != nil { + writeMetricError(w, http.StatusInternalServerError, err.Error()) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "items": rows, + "total": len(rows), + "limit": query.Limit, + "offset": query.Offset, + }) +} + +func parseMetricQuery(r *http.Request) (MetricQuery, error) { + values := r.URL.Query() + limit, err := parseBoundedInt(values.Get("limit"), 50, 1, 1000, "limit") + if err != nil { + return MetricQuery{}, err + } + offset, err := parseBoundedInt(values.Get("offset"), 0, 0, 1_000_000, "offset") + if err != nil { + return MetricQuery{}, err + } + query := MetricQuery{ + VIN: values.Get("vin"), + Protocol: values.Get("protocol"), + MetricKey: values.Get("metricKey"), + DateFrom: values.Get("dateFrom"), + DateTo: values.Get("dateTo"), + Limit: limit, + Offset: offset, + } + if !validDate(query.DateFrom) || !validDate(query.DateTo) { + return MetricQuery{}, errors.New("dateFrom/dateTo must use YYYY-MM-DD") + } + return normalizeMetricQuery(query), nil +} + +func parseBoundedInt(raw string, fallback int, min int, max int, name string) (int, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return fallback, nil + } + value, err := strconv.Atoi(raw) + if err != nil || value < min || value > max { + return 0, fmt.Errorf("%s must be between %d and %d", name, min, max) + } + return value, nil +} + +func validDate(value string) bool { + value = strings.TrimSpace(value) + if value == "" { + return true + } + if len(value) != len("2006-01-02") { + return false + } + for i, r := range value { + switch i { + case 4, 7: + if r != '-' { + return false + } + default: + if r < '0' || r > '9' { + return false + } + } + } + return true +} + +func writeMetricError(w http.ResponseWriter, status int, message string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(map[string]any{"error": message}) +} diff --git a/go/vehicle-gateway/internal/stats/query_test.go b/go/vehicle-gateway/internal/stats/query_test.go new file mode 100644 index 00000000..b9da7df6 --- /dev/null +++ b/go/vehicle-gateway/internal/stats/query_test.go @@ -0,0 +1,100 @@ +package stats + +import ( + "context" + "database/sql" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/DATA-DOG/go-sqlmock" +) + +func TestMetricRepositoryQueriesDailyMetricsWithFilters(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + mock.ExpectQuery("SELECT vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric"). + WithArgs("LKLG7C4E3NA774736", "JT808", "2026-07-01", "2026-07-01", 20, 0). + WillReturnRows(sqlmock.NewRows([]string{ + "vin", "stat_date", "protocol", "metric_key", "metric_value", "metric_unit", + "first_total_mileage_km", "latest_total_mileage_km", "sample_count", "calculation_method", + "created_at", "updated_at", + }).AddRow( + "LKLG7C4E3NA774736", "2026-07-01", "JT808", MetricDailyTotalMileageKM, 12345.6, "km", + 12345.6, 12345.6, 13, "TOTAL_MILEAGE_DIFF", "2026-07-01 22:49:11", "2026-07-01 23:09:36", + )) + + repository := NewMetricRepository(db) + rows, err := repository.Query(context.Background(), MetricQuery{ + VIN: "LKLG7C4E3NA774736", + Protocol: "JT808", + DateFrom: "2026-07-01", + DateTo: "2026-07-01", + Limit: 20, + }) + if err != nil { + t.Fatalf("Query() error = %v", err) + } + if len(rows) != 1 { + t.Fatalf("row count = %d", len(rows)) + } + if rows[0].MetricKey != MetricDailyTotalMileageKM || rows[0].MetricValue != 12345.6 { + t.Fatalf("unexpected row: %#v", rows[0]) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + +func TestMetricHandlerReturnsDailyMetrics(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + mock.ExpectQuery("SELECT vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric"). + WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01", 50, 0). + WillReturnRows(sqlmock.NewRows([]string{ + "vin", "stat_date", "protocol", "metric_key", "metric_value", "metric_unit", + "first_total_mileage_km", "latest_total_mileage_km", "sample_count", "calculation_method", + "created_at", "updated_at", + }).AddRow( + "LB9A32A21R0LS1707", "2020-07-01", "GB32960", MetricDailyMileageKM, 0.0, "km", + 53490.9, 53490.9, 3, "TOTAL_MILEAGE_DIFF", "2026-07-01 22:07:58", "2026-07-01 22:28:25", + )) + + handler := NewMetricHandler(NewMetricRepository(db)) + request := httptest.NewRequest(http.MethodGet, "/api/stats/daily-metrics?vin=LB9A32A21R0LS1707&protocol=GB32960&dateFrom=2020-07-01&dateTo=2020-07-01", nil) + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) + } + body := response.Body.String() + for _, want := range []string{`"vin":"LB9A32A21R0LS1707"`, `"metric_key":"daily_mileage_km"`, `"total":1`} { + if !strings.Contains(body, want) { + t.Fatalf("response missing %s: %s", want, body) + } + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + +func TestMetricHandlerRejectsInvalidPagination(t *testing.T) { + handler := NewMetricHandler(NewMetricRepository(&sql.DB{})) + request := httptest.NewRequest(http.MethodGet, "/api/stats/daily-metrics?limit=2001", nil) + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusBadRequest { + t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) + } +}