feat: add realtime snapshot and location APIs
This commit is contained in:
lingniu
386
go/vehicle-gateway/internal/realtime/mysql_query.go
Normal file
386
go/vehicle-gateway/internal/realtime/mysql_query.go
Normal file
@@ -0,0 +1,386 @@
|
||||
package realtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
type mysqlQueryer interface {
|
||||
QueryContext(context.Context, string, ...any) (*sql.Rows, error)
|
||||
}
|
||||
|
||||
type RealtimeTableQuery struct {
|
||||
Protocol string
|
||||
VIN string
|
||||
Plate string
|
||||
Limit int
|
||||
Offset int
|
||||
}
|
||||
|
||||
type SnapshotRow struct {
|
||||
Protocol string `json:"protocol"`
|
||||
VIN string `json:"vin"`
|
||||
Plate string `json:"plate"`
|
||||
EventTime string `json:"event_time,omitempty"`
|
||||
ReceivedAt string `json:"received_at,omitempty"`
|
||||
EventID string `json:"event_id"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
UpdatedAt string `json:"updated_at"`
|
||||
}
|
||||
|
||||
type LocationRow struct {
|
||||
Protocol string `json:"protocol"`
|
||||
VIN string `json:"vin"`
|
||||
Plate string `json:"plate"`
|
||||
EventTime string `json:"event_time,omitempty"`
|
||||
Latitude float64 `json:"latitude"`
|
||||
Longitude float64 `json:"longitude"`
|
||||
SpeedKMH *float64 `json:"speed_kmh,omitempty"`
|
||||
TotalMileageKM *float64 `json:"total_mileage_km,omitempty"`
|
||||
SOCPercent *float64 `json:"soc_percent,omitempty"`
|
||||
AltitudeM *float64 `json:"altitude_m,omitempty"`
|
||||
DirectionDeg *float64 `json:"direction_deg,omitempty"`
|
||||
AlarmFlag *int64 `json:"alarm_flag,omitempty"`
|
||||
StatusFlag *int64 `json:"status_flag,omitempty"`
|
||||
ReceivedAt string `json:"received_at,omitempty"`
|
||||
EventID string `json:"event_id"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
UpdatedAt string `json:"updated_at"`
|
||||
}
|
||||
|
||||
type SnapshotQueryRepository struct {
|
||||
db mysqlQueryer
|
||||
}
|
||||
|
||||
func NewSnapshotQueryRepository(db mysqlQueryer) *SnapshotQueryRepository {
|
||||
if db == nil {
|
||||
panic("snapshot query db must not be nil")
|
||||
}
|
||||
return &SnapshotQueryRepository{db: db}
|
||||
}
|
||||
|
||||
func (r *SnapshotQueryRepository) Count(ctx context.Context, query RealtimeTableQuery) (int64, error) {
|
||||
sqlText, args := buildRealtimeCountSQL("vehicle_realtime_snapshot", normalizeRealtimeTableQuery(query))
|
||||
return queryCount(ctx, r.db, sqlText, args)
|
||||
}
|
||||
|
||||
func (r *SnapshotQueryRepository) Query(ctx context.Context, query RealtimeTableQuery) ([]SnapshotRow, error) {
|
||||
query = normalizeRealtimeTableQuery(query)
|
||||
sqlText, args := buildRealtimeSelectSQL(
|
||||
"vehicle_realtime_snapshot",
|
||||
"protocol, vin, plate, event_time, received_at, event_id, created_at, updated_at",
|
||||
query,
|
||||
)
|
||||
rows, err := r.db.QueryContext(ctx, sqlText, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := make([]SnapshotRow, 0)
|
||||
for rows.Next() {
|
||||
var row SnapshotRow
|
||||
var eventTime, receivedAt, createdAt, updatedAt scanSQLDateTime
|
||||
if err := rows.Scan(&row.Protocol, &row.VIN, &row.Plate, &eventTime, &receivedAt, &row.EventID, &createdAt, &updatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row.EventTime = eventTime.String
|
||||
row.ReceivedAt = receivedAt.String
|
||||
row.CreatedAt = createdAt.String
|
||||
row.UpdatedAt = updatedAt.String
|
||||
out = append(out, row)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
type LocationQueryRepository struct {
|
||||
db mysqlQueryer
|
||||
}
|
||||
|
||||
func NewLocationQueryRepository(db mysqlQueryer) *LocationQueryRepository {
|
||||
if db == nil {
|
||||
panic("location query db must not be nil")
|
||||
}
|
||||
return &LocationQueryRepository{db: db}
|
||||
}
|
||||
|
||||
func (r *LocationQueryRepository) Count(ctx context.Context, query RealtimeTableQuery) (int64, error) {
|
||||
sqlText, args := buildRealtimeCountSQL("vehicle_realtime_location", normalizeRealtimeTableQuery(query))
|
||||
return queryCount(ctx, r.db, sqlText, args)
|
||||
}
|
||||
|
||||
func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTableQuery) ([]LocationRow, error) {
|
||||
query = normalizeRealtimeTableQuery(query)
|
||||
sqlText, args := buildRealtimeSelectSQL(
|
||||
"vehicle_realtime_location",
|
||||
"protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id, created_at, updated_at",
|
||||
query,
|
||||
)
|
||||
rows, err := r.db.QueryContext(ctx, sqlText, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := make([]LocationRow, 0)
|
||||
for rows.Next() {
|
||||
var row LocationRow
|
||||
var eventTime, receivedAt, createdAt, updatedAt scanSQLDateTime
|
||||
var speed, mileage, soc, altitude, direction sql.NullFloat64
|
||||
var alarm, status sql.NullInt64
|
||||
if err := rows.Scan(
|
||||
&row.Protocol,
|
||||
&row.VIN,
|
||||
&row.Plate,
|
||||
&eventTime,
|
||||
&row.Latitude,
|
||||
&row.Longitude,
|
||||
&speed,
|
||||
&mileage,
|
||||
&soc,
|
||||
&altitude,
|
||||
&direction,
|
||||
&alarm,
|
||||
&status,
|
||||
&receivedAt,
|
||||
&row.EventID,
|
||||
&createdAt,
|
||||
&updatedAt,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row.EventTime = eventTime.String
|
||||
row.SpeedKMH = nullableFloat(speed)
|
||||
row.TotalMileageKM = nullableFloat(mileage)
|
||||
row.SOCPercent = nullableFloat(soc)
|
||||
row.AltitudeM = nullableFloat(altitude)
|
||||
row.DirectionDeg = nullableFloat(direction)
|
||||
row.AlarmFlag = nullableInt(alarm)
|
||||
row.StatusFlag = nullableInt(status)
|
||||
row.ReceivedAt = receivedAt.String
|
||||
row.CreatedAt = createdAt.String
|
||||
row.UpdatedAt = updatedAt.String
|
||||
out = append(out, row)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
type SnapshotQueryHandler struct {
|
||||
repository *SnapshotQueryRepository
|
||||
}
|
||||
|
||||
func NewSnapshotQueryHandler(repository *SnapshotQueryRepository) *SnapshotQueryHandler {
|
||||
if repository == nil {
|
||||
panic("snapshot query repository must not be nil")
|
||||
}
|
||||
return &SnapshotQueryHandler{repository: repository}
|
||||
}
|
||||
|
||||
func (h *SnapshotQueryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
writeError(w, http.StatusMethodNotAllowed, "method not allowed")
|
||||
return
|
||||
}
|
||||
if strings.Trim(r.URL.Path, "/") != "api/realtime/snapshots" {
|
||||
writeError(w, http.StatusNotFound, "route not found")
|
||||
return
|
||||
}
|
||||
query, err := parseRealtimeTableQuery(r)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
total, err := h.repository.Count(r.Context(), query)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
rows, err := h.repository.Query(r.Context(), query)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
writePage(w, rows, total, query)
|
||||
}
|
||||
|
||||
type LocationQueryHandler struct {
|
||||
repository *LocationQueryRepository
|
||||
}
|
||||
|
||||
func NewLocationQueryHandler(repository *LocationQueryRepository) *LocationQueryHandler {
|
||||
if repository == nil {
|
||||
panic("location query repository must not be nil")
|
||||
}
|
||||
return &LocationQueryHandler{repository: repository}
|
||||
}
|
||||
|
||||
func (h *LocationQueryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
writeError(w, http.StatusMethodNotAllowed, "method not allowed")
|
||||
return
|
||||
}
|
||||
if strings.Trim(r.URL.Path, "/") != "api/realtime/locations" {
|
||||
writeError(w, http.StatusNotFound, "route not found")
|
||||
return
|
||||
}
|
||||
query, err := parseRealtimeTableQuery(r)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
total, err := h.repository.Count(r.Context(), query)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
rows, err := h.repository.Query(r.Context(), query)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
writePage(w, rows, total, query)
|
||||
}
|
||||
|
||||
func parseRealtimeTableQuery(r *http.Request) (RealtimeTableQuery, error) {
|
||||
values := r.URL.Query()
|
||||
limit, err := parseRealtimeBoundedInt(values.Get("limit"), 50, 1, 1000, "limit")
|
||||
if err != nil {
|
||||
return RealtimeTableQuery{}, err
|
||||
}
|
||||
offset, err := parseRealtimeBoundedInt(values.Get("offset"), 0, 0, 1_000_000, "offset")
|
||||
if err != nil {
|
||||
return RealtimeTableQuery{}, err
|
||||
}
|
||||
return normalizeRealtimeTableQuery(RealtimeTableQuery{
|
||||
Protocol: values.Get("protocol"),
|
||||
VIN: values.Get("vin"),
|
||||
Plate: values.Get("plate"),
|
||||
Limit: limit,
|
||||
Offset: offset,
|
||||
}), nil
|
||||
}
|
||||
|
||||
func normalizeRealtimeTableQuery(query RealtimeTableQuery) RealtimeTableQuery {
|
||||
query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol))
|
||||
query.VIN = strings.TrimSpace(query.VIN)
|
||||
query.Plate = strings.TrimSpace(query.Plate)
|
||||
if query.Limit <= 0 {
|
||||
query.Limit = 50
|
||||
}
|
||||
return query
|
||||
}
|
||||
|
||||
func buildRealtimeCountSQL(table string, query RealtimeTableQuery) (string, []any) {
|
||||
where, args := buildRealtimeWhere(query)
|
||||
sqlText := "SELECT COUNT(*) FROM " + table
|
||||
if len(where) > 0 {
|
||||
sqlText += " WHERE " + strings.Join(where, " AND ")
|
||||
}
|
||||
return sqlText, args
|
||||
}
|
||||
|
||||
func buildRealtimeSelectSQL(table string, columns string, query RealtimeTableQuery) (string, []any) {
|
||||
where, args := buildRealtimeWhere(query)
|
||||
sqlText := "SELECT " + columns + " FROM " + table
|
||||
if len(where) > 0 {
|
||||
sqlText += " WHERE " + strings.Join(where, " AND ")
|
||||
}
|
||||
sqlText += " ORDER BY updated_at DESC LIMIT ? OFFSET ?"
|
||||
args = append(args, query.Limit, query.Offset)
|
||||
return sqlText, args
|
||||
}
|
||||
|
||||
func buildRealtimeWhere(query RealtimeTableQuery) ([]string, []any) {
|
||||
var where []string
|
||||
var args []any
|
||||
if query.Protocol != "" {
|
||||
where = append(where, "protocol = ?")
|
||||
args = append(args, query.Protocol)
|
||||
}
|
||||
if query.VIN != "" {
|
||||
where = append(where, "vin = ?")
|
||||
args = append(args, query.VIN)
|
||||
}
|
||||
if query.Plate != "" {
|
||||
where = append(where, "plate = ?")
|
||||
args = append(args, query.Plate)
|
||||
}
|
||||
return where, args
|
||||
}
|
||||
|
||||
func queryCount(ctx context.Context, db mysqlQueryer, sqlText string, args []any) (int64, error) {
|
||||
rows, err := db.QueryContext(ctx, sqlText, args...)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var total int64
|
||||
if rows.Next() {
|
||||
if err := rows.Scan(&total); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
}
|
||||
return total, rows.Err()
|
||||
}
|
||||
|
||||
func parseRealtimeBoundedInt(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 writePage(w http.ResponseWriter, items any, total int64, query RealtimeTableQuery) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"items": items,
|
||||
"total": total,
|
||||
"limit": query.Limit,
|
||||
"offset": query.Offset,
|
||||
})
|
||||
}
|
||||
|
||||
func nullableFloat(value sql.NullFloat64) *float64 {
|
||||
if !value.Valid {
|
||||
return nil
|
||||
}
|
||||
return &value.Float64
|
||||
}
|
||||
|
||||
func nullableInt(value sql.NullInt64) *int64 {
|
||||
if !value.Valid {
|
||||
return nil
|
||||
}
|
||||
return &value.Int64
|
||||
}
|
||||
|
||||
type scanSQLDateTime struct {
|
||||
String string
|
||||
}
|
||||
|
||||
func (s *scanSQLDateTime) Scan(value any) error {
|
||||
switch typed := value.(type) {
|
||||
case nil:
|
||||
s.String = ""
|
||||
case time.Time:
|
||||
s.String = typed.Format("2006-01-02 15:04:05.000")
|
||||
case []byte:
|
||||
s.String = strings.TrimSpace(string(typed))
|
||||
case string:
|
||||
s.String = strings.TrimSpace(typed)
|
||||
default:
|
||||
s.String = strings.TrimSpace(fmt.Sprint(typed))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user