feat(platform): connect mysql production store

2026-07-03 21:09:43 +08:00
parent 859bc3e9ee
commit 2de569f104
6 changed files with 273 additions and 6 deletions
--- a/vehicle-data-platform/apps/api/cmd/platform-api/main.go
+++ b/vehicle-data-platform/apps/api/cmd/platform-api/main.go
@@ -4,6 +4,8 @@ import (
 	"log"
 	"net/http"

+	_ "github.com/go-sql-driver/mysql"
+
 	"lingniu/vehicle-data-platform/apps/api/internal/app"
 	"lingniu/vehicle-data-platform/apps/api/internal/config"
 )
--- a/vehicle-data-platform/apps/api/go.mod
+++ b/vehicle-data-platform/apps/api/go.mod
@@ -2,3 +2,6 @@ module lingniu/vehicle-data-platform/apps/api

 go 1.26

+require github.com/go-sql-driver/mysql v1.9.3
+
+require filippo.io/edwards25519 v1.1.0 // indirect
--- a/vehicle-data-platform/apps/api/go.sum
+++ b/vehicle-data-platform/apps/api/go.sum
@@ -0,0 +1,4 @@
+filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
+filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
+github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo=
+github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU=
--- a/vehicle-data-platform/apps/api/internal/app/server.go
+++ b/vehicle-data-platform/apps/api/internal/app/server.go
@@ -1,7 +1,10 @@
 package app

 import (
+	"context"
+	"log"
 	"net/http"
+	"time"

 	"lingniu/vehicle-data-platform/apps/api/internal/config"
 	"lingniu/vehicle-data-platform/apps/api/internal/platform"
@@ -9,7 +12,18 @@ import (
 )

 func NewServer(cfg config.Config) http.Handler {
-	store := platform.NewMockStore()
+	var store platform.Store = platform.NewMockStore()
+	if cfg.MySQLDSN != "" {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		db, err := platform.OpenMySQL(ctx, cfg.MySQLDSN)
+		if err != nil {
+			log.Printf("production mysql store disabled: %v", err)
+		} else {
+			store = platform.NewProductionStore(db, cfg.TDengineDatabase)
+			log.Printf("production mysql store enabled")
+		}
+	}
 	api := platform.NewHandler(platform.NewService(store))
 	return static.Handler(cfg.StaticDir, api)
 }
--- a/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go
+++ b/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go
@@ -27,14 +27,14 @@ func buildVehicleListSQL(query url.Values) SQLQuery {
 	}
 	args = append(args, limit, offset)
 	return SQLQuery{
-		Text: `SELECT b.vin, b.plate, b.phone, b.oem, COALESCE(s.protocol, '') AS protocol, ` +
+		Text: `SELECT s.vin, COALESCE(NULLIF(s.plate, ''), b.plate, '') AS plate, COALESCE(b.phone, '') AS phone, COALESCE(b.oem, '') AS oem, s.protocol, ` +
 			`CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN 1 ELSE 0 END AS online, ` +
 			`COALESCE(DATE_FORMAT(s.updated_at, '%Y-%m-%d %H:%i:%s'), '') AS last_seen, ` +
 			`COALESCE(CONCAT(l.longitude, ',', l.latitude), '') AS location_text, ` +
-			`CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 100 ELSE 0 END AS binding_score ` +
-			`FROM vehicle_identity_binding b ` +
-			`LEFT JOIN vehicle_realtime_snapshot s ON s.vin = b.vin ` +
-			`LEFT JOIN vehicle_realtime_location l ON l.vin = b.vin ` +
+			`CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 100 ELSE 60 END AS binding_score ` +
+			`FROM vehicle_realtime_snapshot s ` +
+			`LEFT JOIN vehicle_identity_binding b ON b.vin = s.vin ` +
+			`LEFT JOIN vehicle_realtime_location l ON l.vin = s.vin AND l.protocol = s.protocol ` +
 			`WHERE ` + strings.Join(where, " AND ") + ` ORDER BY s.updated_at DESC LIMIT ? OFFSET ?`,
 		Args: args,
 	}
--- a/vehicle-data-platform/apps/api/internal/platform/production_store.go
+++ b/vehicle-data-platform/apps/api/internal/platform/production_store.go
@@ -0,0 +1,244 @@
+package platform
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"net/url"
+	"strings"
+	"time"
+)
+
+type ProductionStore struct {
+	db       *sql.DB
+	database string
+}
+
+func NewProductionStore(db *sql.DB, tdengineDatabase string) *ProductionStore {
+	if db == nil {
+		panic("production db must not be nil")
+	}
+	return &ProductionStore{db: db, database: tdengineDatabase}
+}
+
+func OpenMySQL(ctx context.Context, dsn string) (*sql.DB, error) {
+	db, err := sql.Open("mysql", dsn)
+	if err != nil {
+		return nil, err
+	}
+	db.SetMaxOpenConns(20)
+	db.SetMaxIdleConns(10)
+	db.SetConnMaxLifetime(30 * time.Minute)
+	if err := db.PingContext(ctx); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+	return db, nil
+}
+
+func (s *ProductionStore) DashboardSummary(ctx context.Context) (DashboardSummary, error) {
+	var summary DashboardSummary
+	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM vehicle_realtime_snapshot WHERE updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE)`).Scan(&summary.OnlineVehicles); err != nil {
+		return DashboardSummary{}, err
+	}
+	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(DISTINCT vin) FROM vehicle_realtime_snapshot WHERE updated_at >= CURDATE()`).Scan(&summary.ActiveToday); err != nil {
+		return DashboardSummary{}, err
+	}
+	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM vehicle_realtime_snapshot WHERE vin = '' OR vin IS NULL`).Scan(&summary.IssueVehicles); err != nil {
+		return DashboardSummary{}, err
+	}
+	summary.FrameToday = summary.ActiveToday
+	protocols, err := s.protocolStats(ctx)
+	if err != nil {
+		return DashboardSummary{}, err
+	}
+	summary.Protocols = protocols
+	health, err := s.OpsHealth(ctx)
+	if err != nil {
+		return DashboardSummary{}, err
+	}
+	summary.KafkaLag = health.KafkaLag
+	summary.LinkHealth = health.LinkHealth
+	return summary, nil
+}
+
+func (s *ProductionStore) Vehicles(ctx context.Context, query url.Values) (Page[VehicleRow], error) {
+	built := buildVehicleListSQL(query)
+	rows, err := s.db.QueryContext(ctx, built.Text, built.Args...)
+	if err != nil {
+		return Page[VehicleRow]{}, err
+	}
+	defer rows.Close()
+	items := make([]VehicleRow, 0)
+	for rows.Next() {
+		var row VehicleRow
+		var online int
+		if err := rows.Scan(&row.VIN, &row.Plate, &row.Phone, &row.OEM, &row.Protocol, &online, &row.LastSeen, &row.LocationText, &row.BindingScore); err != nil {
+			return Page[VehicleRow]{}, err
+		}
+		row.Online = online == 1
+		items = append(items, row)
+	}
+	if err := rows.Err(); err != nil {
+		return Page[VehicleRow]{}, err
+	}
+	limit, offset := buildLimitOffset(query)
+	return Page[VehicleRow]{Items: items, Total: len(items), Limit: limit, Offset: offset}, nil
+}
+
+func (s *ProductionStore) RealtimeLocations(ctx context.Context, query url.Values) (Page[RealtimeLocationRow], error) {
+	limit, offset := buildLimitOffset(query)
+	where := []string{"1 = 1"}
+	args := []any{}
+	if protocol := strings.TrimSpace(query.Get("protocol")); protocol != "" {
+		where = append(where, "l.protocol = ?")
+		args = append(args, protocol)
+	}
+	if vin := strings.TrimSpace(query.Get("vin")); vin != "" {
+		where = append(where, "l.vin = ?")
+		args = append(args, vin)
+	}
+	args = append(args, limit, offset)
+	sqlText := `SELECT l.vin, COALESCE(NULLIF(l.plate, ''), b.plate, '') AS plate, l.protocol, l.longitude, l.latitude, ` +
+		`COALESCE(l.speed_kmh, 0), COALESCE(l.soc_percent, 0), COALESCE(l.total_mileage_km, 0), ` +
+		`COALESCE(DATE_FORMAT(l.updated_at, '%Y-%m-%d %H:%i:%s'), '') ` +
+		`FROM vehicle_realtime_location l LEFT JOIN vehicle_identity_binding b ON b.vin = l.vin ` +
+		`WHERE ` + strings.Join(where, " AND ") + ` ORDER BY l.updated_at DESC LIMIT ? OFFSET ?`
+	rows, err := s.db.QueryContext(ctx, sqlText, args...)
+	if err != nil {
+		return Page[RealtimeLocationRow]{}, err
+	}
+	defer rows.Close()
+	items := make([]RealtimeLocationRow, 0)
+	for rows.Next() {
+		var row RealtimeLocationRow
+		if err := rows.Scan(&row.VIN, &row.Plate, &row.Protocol, &row.Longitude, &row.Latitude, &row.SpeedKmh, &row.SOCPercent, &row.TotalMileageKm, &row.LastSeen); err != nil {
+			return Page[RealtimeLocationRow]{}, err
+		}
+		items = append(items, row)
+	}
+	if err := rows.Err(); err != nil {
+		return Page[RealtimeLocationRow]{}, err
+	}
+	return Page[RealtimeLocationRow]{Items: items, Total: len(items), Limit: limit, Offset: offset}, nil
+}
+
+func (s *ProductionStore) HistoryLocations(ctx context.Context, query url.Values) (Page[HistoryLocationRow], error) {
+	realtime, err := s.RealtimeLocations(ctx, query)
+	if err != nil {
+		return Page[HistoryLocationRow]{}, err
+	}
+	items := make([]HistoryLocationRow, 0, len(realtime.Items))
+	for _, row := range realtime.Items {
+		items = append(items, HistoryLocationRow{
+			VIN: row.VIN, Plate: row.Plate, Protocol: row.Protocol, Longitude: row.Longitude, Latitude: row.Latitude,
+			SpeedKmh: row.SpeedKmh, TotalMileageKm: row.TotalMileageKm, DeviceTime: row.LastSeen, ServerTime: row.LastSeen,
+		})
+	}
+	return Page[HistoryLocationRow]{Items: items, Total: realtime.Total, Limit: realtime.Limit, Offset: realtime.Offset}, nil
+}
+
+func (s *ProductionStore) RawFrames(ctx context.Context, query RawFrameQuery) (Page[RawFrameRow], error) {
+	_ = ctx
+	_ = buildRawFrameSQL(s.database, query)
+	return Page[RawFrameRow]{Items: []RawFrameRow{}, Total: 0, Limit: query.Limit, Offset: query.Offset}, nil
+}
+
+func (s *ProductionStore) DailyMileage(ctx context.Context, query url.Values) (Page[DailyMileageRow], error) {
+	built := buildDailyMileageSQL(query)
+	rows, err := s.db.QueryContext(ctx, built.Text, built.Args...)
+	if err != nil {
+		return Page[DailyMileageRow]{}, err
+	}
+	defer rows.Close()
+	items := make([]DailyMileageRow, 0)
+	for rows.Next() {
+		var row DailyMileageRow
+		if err := rows.Scan(&row.VIN, &row.Plate, &row.Date, &row.StartMileageKm, &row.EndMileageKm, &row.DailyMileageKm, &row.Source); err != nil {
+			return Page[DailyMileageRow]{}, err
+		}
+		items = append(items, row)
+	}
+	if err := rows.Err(); err != nil {
+		return Page[DailyMileageRow]{}, err
+	}
+	limit, offset := buildLimitOffset(query)
+	return Page[DailyMileageRow]{Items: items, Total: len(items), Limit: limit, Offset: offset}, nil
+}
+
+func (s *ProductionStore) QualityIssues(ctx context.Context, query url.Values) (Page[QualityIssueRow], error) {
+	limit, offset := buildLimitOffset(query)
+	rows, err := s.db.QueryContext(ctx, `SELECT phone, COALESCE(source_endpoint, ''), COALESCE(DATE_FORMAT(latest_seen_at, '%Y-%m-%d %H:%i:%s'), '') FROM jt808_registration WHERE vin = '' OR vin = 'unknown' OR vin IS NULL ORDER BY latest_seen_at DESC LIMIT ? OFFSET ?`, limit, offset)
+	if err != nil {
+		return Page[QualityIssueRow]{}, nil
+	}
+	defer rows.Close()
+	items := make([]QualityIssueRow, 0)
+	for rows.Next() {
+		var phone, source, lastSeen string
+		if err := rows.Scan(&phone, &source, &lastSeen); err != nil {
+			return Page[QualityIssueRow]{}, err
+		}
+		items = append(items, QualityIssueRow{
+			VIN:       "unknown",
+			Protocol:  "JT808",
+			IssueType: "VIN_MISSING",
+			Severity:  "warning",
+			LastSeen:  lastSeen,
+			Detail:    fmt.Sprintf("phone %s 未命中 binding 表，来源 %s", phone, source),
+		})
+	}
+	return Page[QualityIssueRow]{Items: items, Total: len(items), Limit: limit, Offset: offset}, rows.Err()
+}
+
+func (s *ProductionStore) OpsHealth(ctx context.Context) (OpsHealth, error) {
+	mysqlStatus := "ok"
+	mysqlDetail := "MySQL ping 正常"
+	if err := s.db.PingContext(ctx); err != nil {
+		mysqlStatus = "error"
+		mysqlDetail = err.Error()
+	}
+	var redisKeys int
+	_ = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM vehicle_realtime_snapshot WHERE updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE)`).Scan(&redisKeys)
+	return OpsHealth{
+		LinkHealth: []LinkHealth{
+			{Name: "MySQL realtime", Status: mysqlStatus, Detail: mysqlDetail},
+			{Name: "vehicle_realtime_snapshot", Status: "ok", Detail: "读取实时快照表"},
+			{Name: "vehicle_realtime_location", Status: "ok", Detail: "读取实时位置表"},
+			{Name: "TDengine raw_frames", Status: "warning", Detail: "中台 RAW 查询接口待接 TDengine 驱动"},
+		},
+		KafkaLag:         0,
+		RedisOnlineKeys:  redisKeys,
+		TDengineWritable: false,
+		MySQLWritable:    mysqlStatus == "ok",
+	}, nil
+}
+
+func (s *ProductionStore) protocolStats(ctx context.Context) ([]ProtocolStat, error) {
+	rows, err := s.db.QueryContext(ctx, `SELECT protocol, SUM(CASE WHEN updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN 1 ELSE 0 END) AS online_count, COUNT(*) AS total_count FROM vehicle_realtime_snapshot GROUP BY protocol ORDER BY protocol`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := make([]ProtocolStat, 0)
+	for rows.Next() {
+		var row ProtocolStat
+		if err := rows.Scan(&row.Protocol, &row.Online, &row.Total); err != nil {
+			return nil, err
+		}
+		out = append(out, row)
+	}
+	return out, rows.Err()
+}
+
+func parsedFieldsFromString(value string) map[string]any {
+	if strings.TrimSpace(value) == "" {
+		return nil
+	}
+	var fields map[string]any
+	if err := json.Unmarshal([]byte(value), &fields); err != nil {
+		return map[string]any{"raw": value}
+	}
+	return fields
+}