shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(stats): slim final daily mileage table

Browse Source
This commit is contained in:
lingniu
2026-07-08 16:58:25 +08:00
parent da669fae94
commit 536b0bfee6
7 changed files with 109 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
go/vehicle-gateway/internal/stats/daily_metric.go
View File

@@ -72,7 +72,7 @@ func (w *Writer) EnsureSchema(ctx context.Context) error {
} }
} }
for _, statement := range DailyMileageAlterSQL { for _, statement := range DailyMileageAlterSQL {
if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) { if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isIgnorableSchemaChangeError(err) {
return err return err
} }
} }
@@ -284,12 +284,17 @@ func sourceKey(env envelope.FrameEnvelope) string {
return SourceKey(env.Protocol, env.Phone, env.DeviceID, NormalizeSourceIP(env.SourceEndpoint)) return SourceKey(env.Protocol, env.Phone, env.DeviceID, NormalizeSourceIP(env.SourceEndpoint))
} }
func isDuplicateColumnError(err error) bool { func isIgnorableSchemaChangeError(err error) bool {
if err == nil { if err == nil {
return false return false
} }
text := strings.ToLower(err.Error()) text := strings.ToLower(err.Error())
return strings.Contains(text, "duplicate column") || strings.Contains(text, "1060") return strings.Contains(text, "duplicate column") ||
strings.Contains(text, "1060") ||
strings.Contains(text, "duplicate key name") ||
strings.Contains(text, "1061") ||
strings.Contains(text, "can't drop") ||
strings.Contains(text, "1091")
} }
type mileageFieldMapping struct { type mileageFieldMapping struct {

30
go/vehicle-gateway/internal/stats/daily_metric_test.go
View File

@@ -559,7 +559,7 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
t.Fatalf("Append() error = %v", err) t.Fatalf("Append() error = %v", err)
} }
if len(exec.calls) != 12 { if len(exec.calls) != 16 {
t.Fatalf("exec calls = %d", len(exec.calls)) t.Fatalf("exec calls = %d", len(exec.calls))
} }
for i, want := range []string{ for i, want := range []string{
@@ -571,18 +571,23 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
t.Fatalf("schema call %d = %s, want %s", i, exec.calls[i].query, want) t.Fatalf("schema call %d = %s, want %s", i, exec.calls[i].query, want)
} }
} }
for _, column := range []string{"vehicle_key", "id BIGINT", "AUTO_INCREMENT", "created_at"} { for _, column := range []string{"vehicle_key", "AUTO_INCREMENT", "created_at", "first_total_mileage_km", "trusted_source_key", "trusted_phone", "trusted_source_endpoint", "sample_count"} {
if strings.Contains(exec.calls[2].query, column) { if strings.Contains(exec.calls[2].query, column) {
t.Fatalf("schema should not include %s: %s", column, exec.calls[2].query) t.Fatalf("schema should not include %s: %s", column, exec.calls[2].query)
} }
} }
for _, column := range []string{"source_id BIGINT NULL", "daily_mileage_km", "latest_total_mileage_km", "KEY idx_source_id (source_id)"} {
if !strings.Contains(exec.calls[2].query, column) {
t.Fatalf("schema should include %s: %s", column, exec.calls[2].query)
}
}
if !strings.Contains(exec.calls[2].query, "PRIMARY KEY (vin, stat_date, protocol)") { if !strings.Contains(exec.calls[2].query, "PRIMARY KEY (vin, stat_date, protocol)") {
t.Fatalf("daily mileage table should key by vin/stat_date/protocol: %s", exec.calls[2].query) t.Fatalf("daily mileage table should key by vin/stat_date/protocol: %s", exec.calls[2].query)
} }
if strings.Contains(exec.calls[2].query, "KEY idx_vin (vin)") { if strings.Contains(exec.calls[2].query, "KEY idx_vin (vin)") {
t.Fatalf("daily mileage table should not keep redundant vin index covered by the primary key: %s", exec.calls[2].query) t.Fatalf("daily mileage table should not keep redundant vin index covered by the primary key: %s", exec.calls[2].query)
} }
projectCall := exec.calls[9] projectCall := exec.calls[13]
if !strings.Contains(projectCall.query, "INSERT INTO vehicle_daily_mileage") { if !strings.Contains(projectCall.query, "INSERT INTO vehicle_daily_mileage") {
t.Fatalf("unexpected project sql: %s", projectCall.query) t.Fatalf("unexpected project sql: %s", projectCall.query)
} }
@@ -592,21 +597,26 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
if strings.Contains(projectCall.query, "metric_key") || strings.Contains(projectCall.query, "metric_unit") { if strings.Contains(projectCall.query, "metric_key") || strings.Contains(projectCall.query, "metric_unit") {
t.Fatalf("daily mileage upsert should not use generic metric columns: %s", projectCall.query) t.Fatalf("daily mileage upsert should not use generic metric columns: %s", projectCall.query)
} }
if !strings.Contains(projectCall.query, "LEFT JOIN vehicle_data_source ds") { if !strings.Contains(projectCall.query, "JOIN vehicle_data_source ds") {
t.Fatalf("project sql should join data source: %s", projectCall.query) t.Fatalf("project sql should join data source: %s", projectCall.query)
} }
if !strings.Contains(projectCall.query, "trusted_source_key") { if !strings.Contains(projectCall.query, "ds.id") {
t.Fatalf("upsert should track trusted source: %s", projectCall.query) t.Fatalf("project sql should write source id: %s", projectCall.query)
}
for _, forbidden := range []string{"trusted_source_key", "trusted_phone", "trusted_source_endpoint", "first_total_mileage_km", "sample_count)"} {
if strings.Contains(projectCall.query, forbidden) {
t.Fatalf("project sql should not write final-table field %q: %s", forbidden, projectCall.query)
}
} }
if got := projectCall.args[0]; got != "LNBVIN00000000002" { if got := projectCall.args[0]; got != "LNBVIN00000000002" {
t.Fatalf("first upsert arg should be vin, got %#v", got) t.Fatalf("first upsert arg should be vin, got %#v", got)
} }
if !strings.Contains(exec.calls[6].query, "INSERT INTO vehicle_data_source") { if !strings.Contains(exec.calls[10].query, "INSERT INTO vehicle_data_source") {
t.Fatalf("source upsert query missing: %s", exec.calls[6].query) t.Fatalf("source upsert query missing: %s", exec.calls[10].query)
} }
if !strings.Contains(exec.calls[7].query, "INSERT INTO vehicle_daily_mileage_source") { if !strings.Contains(exec.calls[11].query, "INSERT INTO vehicle_daily_mileage_source") {
t.Fatalf("candidate upsert query missing: %s", exec.calls[7].query) t.Fatalf("candidate upsert query missing: %s", exec.calls[11].query)
} }
} }

49
go/vehicle-gateway/internal/stats/query.go
View File

@@ -27,17 +27,13 @@ type MetricQuery struct {
} }
type MetricRow struct { type MetricRow struct {
VIN string `json:"vin"` VIN string `json:"vin"`
StatDate string `json:"stat_date"` StatDate string `json:"stat_date"`
Protocol string `json:"protocol"` Protocol string `json:"protocol"`
DailyMileageKM float64 `json:"daily_mileage_km"` SourceID *int64 `json:"source_id,omitempty"`
FirstTotalMileageKM *float64 `json:"first_total_mileage_km,omitempty"` DailyMileageKM float64 `json:"daily_mileage_km"`
LatestTotalMileageKM *float64 `json:"latest_total_mileage_km,omitempty"` LatestTotalMileageKM *float64 `json:"latest_total_mileage_km,omitempty"`
TrustedSourceKey *string `json:"trusted_source_key,omitempty"` UpdatedAt string `json:"updated_at"`
TrustedPhone *string `json:"trusted_phone,omitempty"`
TrustedSourceEndpoint *string `json:"trusted_source_endpoint,omitempty"`
SampleCount int64 `json:"sample_count"`
UpdatedAt string `json:"updated_at"`
} }
type MetricRepository struct { type MetricRepository struct {
@@ -65,43 +61,27 @@ func (r *MetricRepository) Query(ctx context.Context, query MetricQuery) ([]Metr
var row MetricRow var row MetricRow
var statDate scanDate var statDate scanDate
var updatedAt scanDateTime var updatedAt scanDateTime
var first sql.NullFloat64 var sourceID sql.NullInt64
var latest sql.NullFloat64 var latest sql.NullFloat64
var trustedSourceKey sql.NullString
var trustedPhone sql.NullString
var trustedSourceEndpoint sql.NullString
if err := rows.Scan( if err := rows.Scan(
&row.VIN, &row.VIN,
&statDate, &statDate,
&row.Protocol, &row.Protocol,
&sourceID,
&row.DailyMileageKM, &row.DailyMileageKM,
&first,
&latest, &latest,
&trustedSourceKey,
&trustedPhone,
&trustedSourceEndpoint,
&row.SampleCount,
&updatedAt, &updatedAt,
); err != nil { ); err != nil {
return nil, err return nil, err
} }
row.StatDate = statDate.String row.StatDate = statDate.String
row.UpdatedAt = updatedAt.String row.UpdatedAt = updatedAt.String
if first.Valid { if sourceID.Valid {
row.FirstTotalMileageKM = &first.Float64 row.SourceID = &sourceID.Int64
} }
if latest.Valid { if latest.Valid {
row.LatestTotalMileageKM = &latest.Float64 row.LatestTotalMileageKM = &latest.Float64
} }
if trustedSourceKey.Valid {
row.TrustedSourceKey = &trustedSourceKey.String
}
if trustedPhone.Valid {
row.TrustedPhone = &trustedPhone.String
}
if trustedSourceEndpoint.Valid {
row.TrustedSourceEndpoint = &trustedSourceEndpoint.String
}
out = append(out, row) out = append(out, row)
} }
return out, rows.Err() return out, rows.Err()
@@ -138,10 +118,9 @@ func normalizeMetricQuery(query MetricQuery) MetricQuery {
func buildMetricSQL(query MetricQuery) (string, []any) { func buildMetricSQL(query MetricQuery) (string, []any) {
where, args := buildMetricWhere(query) where, args := buildMetricWhere(query)
sqlText := `SELECT vin, stat_date, protocol, daily_mileage_km, sqlText := `SELECT vin, stat_date, protocol, source_id,
first_total_mileage_km, latest_total_mileage_km, daily_mileage_km, latest_total_mileage_km, updated_at
trusted_source_key, trusted_phone, trusted_source_endpoint, FROM vehicle_daily_mileage`
sample_count, updated_at FROM vehicle_daily_mileage`
if len(where) > 0 { if len(where) > 0 {
sqlText += " WHERE " + strings.Join(where, " AND ") sqlText += " WHERE " + strings.Join(where, " AND ")
} }

79
go/vehicle-gateway/internal/stats/query_test.go
View File

@@ -12,25 +12,22 @@ import (
"github.com/DATA-DOG/go-sqlmock" "github.com/DATA-DOG/go-sqlmock"
) )
const dailyMetricSelectPattern = "SELECT vin, stat_date, protocol, source_id, daily_mileage_km, latest_total_mileage_km, updated_at FROM vehicle_daily_mileage"
func TestMetricRepositoryQueriesDailyMetricsWithFilters(t *testing.T) { func TestMetricRepositoryQueriesDailyMetricsWithFilters(t *testing.T) {
db, mock, err := sqlmock.New() db, mock, err := sqlmock.New()
if err != nil { if err != nil {
t.Fatalf("sqlmock.New() error = %v", err) t.Fatalf("sqlmock.New() error = %v", err)
} }
defer db.Close() defer db.Close()
mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count, updated_at FROM vehicle_daily_mileage"). mock.ExpectQuery(dailyMetricSelectPattern).
WithArgs("LKLG7C4E3NA774736", "JT808", "2026-07-01", "2026-07-01", 20, 0). WithArgs("LKLG7C4E3NA774736", "JT808", "2026-07-01", "2026-07-01", 20, 0).
WillReturnRows(sqlmock.NewRows([]string{ WillReturnRows(sqlmock.NewRows([]string{
"vin", "stat_date", "protocol", "daily_mileage_km", "vin", "stat_date", "protocol", "source_id", "daily_mileage_km",
"first_total_mileage_km", "latest_total_mileage_km", "latest_total_mileage_km", "updated_at",
"trusted_source_key", "trusted_phone", "trusted_source_endpoint",
"sample_count",
"updated_at",
}).AddRow( }).AddRow(
"LKLG7C4E3NA774736", time.Date(2026, 7, 1, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), "JT808", 12.3, "LKLG7C4E3NA774736", time.Date(2026, 7, 1, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), "JT808", 3, 12.3,
12345.6, 12357.9, 12357.9, time.Date(2026, 7, 1, 23, 9, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
nil, nil, nil,
13, time.Date(2026, 7, 1, 23, 9, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
)) ))
repository := NewMetricRepository(db) repository := NewMetricRepository(db)
@@ -50,6 +47,9 @@ func TestMetricRepositoryQueriesDailyMetricsWithFilters(t *testing.T) {
if rows[0].DailyMileageKM != 12.3 { if rows[0].DailyMileageKM != 12.3 {
t.Fatalf("unexpected row: %#v", rows[0]) t.Fatalf("unexpected row: %#v", rows[0])
} }
if rows[0].SourceID == nil || *rows[0].SourceID != 3 {
t.Fatalf("source id = %#v", rows[0].SourceID)
}
if rows[0].StatDate != "2026-07-01" || rows[0].UpdatedAt != "2026-07-01 23:09:36" { if rows[0].StatDate != "2026-07-01" || rows[0].UpdatedAt != "2026-07-01 23:09:36" {
t.Fatalf("unexpected time formatting: %#v", rows[0]) t.Fatalf("unexpected time formatting: %#v", rows[0])
} }
@@ -58,23 +58,19 @@ func TestMetricRepositoryQueriesDailyMetricsWithFilters(t *testing.T) {
} }
} }
func TestMetricQueryReturnsSelectedSourceFields(t *testing.T) { func TestMetricQueryReturnsSelectedSourceID(t *testing.T) {
db, mock, err := sqlmock.New() db, mock, err := sqlmock.New()
if err != nil { if err != nil {
t.Fatalf("sqlmock.New() error = %v", err) t.Fatalf("sqlmock.New() error = %v", err)
} }
defer db.Close() defer db.Close()
mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count, updated_at FROM vehicle_daily_mileage"). mock.ExpectQuery(dailyMetricSelectPattern).
WillReturnRows(sqlmock.NewRows([]string{ WillReturnRows(sqlmock.NewRows([]string{
"vin", "stat_date", "protocol", "daily_mileage_km", "vin", "stat_date", "protocol", "source_id", "daily_mileage_km",
"first_total_mileage_km", "latest_total_mileage_km", "latest_total_mileage_km", "updated_at",
"trusted_source_key", "trusted_phone", "trusted_source_endpoint",
"sample_count", "updated_at",
}).AddRow( }).AddRow(
"LA9GG64L7PBAF4001", "2026-07-08", "JT808", 23.1, "LA9GG64L7PBAF4001", "2026-07-08", "JT808", 2, 23.1,
4100.8, 4123.9, 4123.9, "2026-07-08 13:30:57",
"JT808:13307765812@115.231.168.135", "13307765812", "115.231.168.135:20215",
1, "2026-07-08 13:30:57",
)) ))
repo := NewMetricRepository(db) repo := NewMetricRepository(db)
@@ -86,8 +82,8 @@ func TestMetricQueryReturnsSelectedSourceFields(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("Query() error = %v", err) t.Fatalf("Query() error = %v", err)
} }
if got[0].TrustedSourceKey == nil || *got[0].TrustedSourceKey != "JT808:13307765812@115.231.168.135" { if got[0].SourceID == nil || *got[0].SourceID != 2 {
t.Fatalf("trusted source = %#v", got[0].TrustedSourceKey) t.Fatalf("source id = %#v", got[0].SourceID)
} }
} }
@@ -100,19 +96,14 @@ func TestMetricHandlerReturnsDailyMetrics(t *testing.T) {
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_daily_mileage"). mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_daily_mileage").
WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01"). WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(42)) WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(42))
mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count, updated_at FROM vehicle_daily_mileage"). mock.ExpectQuery(dailyMetricSelectPattern).
WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01", 50, 0). WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01", 50, 0).
WillReturnRows(sqlmock.NewRows([]string{ WillReturnRows(sqlmock.NewRows([]string{
"vin", "stat_date", "protocol", "daily_mileage_km", "vin", "stat_date", "protocol", "source_id", "daily_mileage_km",
"first_total_mileage_km", "latest_total_mileage_km", "latest_total_mileage_km", "updated_at",
"trusted_source_key", "trusted_phone", "trusted_source_endpoint",
"sample_count",
"updated_at",
}).AddRow( }).AddRow(
"LB9A32A21R0LS1707", "2020-07-01", "GB32960", 0.0, "LB9A32A21R0LS1707", "2020-07-01", "GB32960", 1, 0.0,
53490.9, 53490.9, 53490.9, "2026-07-01 22:28:25",
nil, nil, nil,
3, "2026-07-01 22:28:25",
)) ))
handler := NewMetricHandler(NewMetricRepository(db)) handler := NewMetricHandler(NewMetricRepository(db))
@@ -144,14 +135,11 @@ func TestMetricHandlerReturnsEmptyItemsArrayWhenNoRows(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err) t.Fatalf("sqlmock.New() error = %v", err)
} }
defer db.Close() defer db.Close()
mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count, updated_at FROM vehicle_daily_mileage"). mock.ExpectQuery(dailyMetricSelectPattern).
WithArgs("YUTONG_MQTT", 50, 0). WithArgs("YUTONG_MQTT", 50, 0).
WillReturnRows(sqlmock.NewRows([]string{ WillReturnRows(sqlmock.NewRows([]string{
"vin", "stat_date", "protocol", "daily_mileage_km", "vin", "stat_date", "protocol", "source_id", "daily_mileage_km",
"first_total_mileage_km", "latest_total_mileage_km", "latest_total_mileage_km", "updated_at",
"trusted_source_key", "trusted_phone", "trusted_source_endpoint",
"sample_count",
"updated_at",
})) }))
handler := NewMetricHandler(NewMetricRepository(db)) handler := NewMetricHandler(NewMetricRepository(db))
@@ -178,19 +166,14 @@ func TestMetricHandlerSkipsTotalCountByDefault(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err) t.Fatalf("sqlmock.New() error = %v", err)
} }
defer db.Close() defer db.Close()
mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count, updated_at FROM vehicle_daily_mileage"). mock.ExpectQuery(dailyMetricSelectPattern).
WithArgs("JT808", 1, 0). WithArgs("JT808", 1, 0).
WillReturnRows(sqlmock.NewRows([]string{ WillReturnRows(sqlmock.NewRows([]string{
"vin", "stat_date", "protocol", "daily_mileage_km", "vin", "stat_date", "protocol", "source_id", "daily_mileage_km",
"first_total_mileage_km", "latest_total_mileage_km", "latest_total_mileage_km", "updated_at",
"trusted_source_key", "trusted_phone", "trusted_source_endpoint",
"sample_count",
"updated_at",
}).AddRow( }).AddRow(
"LKLG7C4E3NA774736", "2026-07-02", "JT808", 12.3, "LKLG7C4E3NA774736", "2026-07-02", "JT808", 3, 12.3,
8792.8, 8805.1, 8805.1, "2026-07-02 23:59:59",
nil, nil, nil,
30, "2026-07-02 23:59:59",
)) ))
handler := NewMetricHandler(NewMetricRepository(db)) handler := NewMetricHandler(NewMetricRepository(db))

19
go/vehicle-gateway/internal/stats/schema.go
View File

@@ -4,23 +4,24 @@ const DailyMileageTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_mileage (
vin VARCHAR(32) NOT NULL, vin VARCHAR(32) NOT NULL,
stat_date DATE NOT NULL, stat_date DATE NOT NULL,
protocol VARCHAR(32) NOT NULL, protocol VARCHAR(32) NOT NULL,
source_id BIGINT NULL,
daily_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0, daily_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0,
first_total_mileage_km DECIMAL(18,3) NULL,
latest_total_mileage_km DECIMAL(18,3) NULL, latest_total_mileage_km DECIMAL(18,3) NULL,
trusted_source_key VARCHAR(256) NULL,
trusted_phone VARCHAR(32) NULL,
trusted_source_endpoint VARCHAR(128) NULL,
sample_count BIGINT NOT NULL DEFAULT 0,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (vin, stat_date, protocol), PRIMARY KEY (vin, stat_date, protocol),
KEY idx_stat_date (stat_date), KEY idx_stat_date (stat_date),
KEY idx_protocol_date (protocol, stat_date) KEY idx_protocol_date (protocol, stat_date),
KEY idx_source_id (source_id)
)` )`
var DailyMileageAlterSQL = []string{ var DailyMileageAlterSQL = []string{
"ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_source_key VARCHAR(256) NULL", "ALTER TABLE vehicle_daily_mileage ADD COLUMN source_id BIGINT NULL AFTER protocol",
"ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_phone VARCHAR(32) NULL", "ALTER TABLE vehicle_daily_mileage ADD KEY idx_source_id (source_id)",
"ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_source_endpoint VARCHAR(128) NULL", "ALTER TABLE vehicle_daily_mileage DROP COLUMN first_total_mileage_km",
"ALTER TABLE vehicle_daily_mileage DROP COLUMN trusted_source_key",
"ALTER TABLE vehicle_daily_mileage DROP COLUMN trusted_phone",
"ALTER TABLE vehicle_daily_mileage DROP COLUMN trusted_source_endpoint",
"ALTER TABLE vehicle_daily_mileage DROP COLUMN sample_count",
} }
const DataSourceTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_data_source ( const DataSourceTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_data_source (

30
go/vehicle-gateway/internal/stats/source_mileage.go
View File

@@ -200,42 +200,32 @@ WHERE vin = ? AND stat_date = ? AND protocol = ?
const projectDailyMileageSQL = ` const projectDailyMileageSQL = `
INSERT INTO vehicle_daily_mileage INSERT INTO vehicle_daily_mileage
(vin, stat_date, protocol, daily_mileage_km, (vin, stat_date, protocol, source_id, daily_mileage_km, latest_total_mileage_km)
first_total_mileage_km, latest_total_mileage_km,
trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count)
SELECT SELECT
s.vin, s.vin,
s.stat_date, s.stat_date,
s.protocol, s.protocol,
ds.id,
s.daily_mileage_km, s.daily_mileage_km,
s.first_total_mileage_km, s.latest_total_mileage_km
s.latest_total_mileage_km,
s.source_key,
s.phone,
ds.latest_source_endpoint,
s.sample_count
FROM vehicle_daily_mileage_source s FROM vehicle_daily_mileage_source s
LEFT JOIN vehicle_data_source ds JOIN vehicle_data_source ds
ON ds.protocol = s.protocol AND ds.source_ip = s.source_ip ON ds.protocol = s.protocol AND ds.source_ip = s.source_ip
WHERE s.vin = ? WHERE s.vin = ?
AND s.stat_date = ? AND s.stat_date = ?
AND s.protocol = ? AND s.protocol = ?
AND s.quality_status = '` + QualityOK + `' AND s.quality_status = '` + QualityOK + `'
AND s.daily_mileage_km BETWEEN 0 AND ? AND s.daily_mileage_km BETWEEN 0 AND ?
AND COALESCE(ds.enabled, 1) = 1 AND ds.enabled = 1
ORDER BY COALESCE(ds.trust_priority, 100), ORDER BY ds.trust_priority,
s.sample_count DESC, s.sample_count DESC,
s.latest_event_time DESC, s.latest_event_time DESC,
s.source_key ASC s.source_key ASC
LIMIT 1 LIMIT 1
ON DUPLICATE KEY UPDATE ON DUPLICATE KEY UPDATE
source_id = VALUES(source_id),
daily_mileage_km = VALUES(daily_mileage_km), daily_mileage_km = VALUES(daily_mileage_km),
first_total_mileage_km = VALUES(first_total_mileage_km),
latest_total_mileage_km = VALUES(latest_total_mileage_km), latest_total_mileage_km = VALUES(latest_total_mileage_km),
trusted_source_key = VALUES(trusted_source_key),
trusted_phone = VALUES(trusted_phone),
trusted_source_endpoint = VALUES(trusted_source_endpoint),
sample_count = VALUES(sample_count),
updated_at = CURRENT_TIMESTAMP updated_at = CURRENT_TIMESTAMP
` `
@@ -248,15 +238,15 @@ JOIN (
s2.stat_date, s2.stat_date,
s2.protocol s2.protocol
FROM vehicle_daily_mileage_source s2 FROM vehicle_daily_mileage_source s2
LEFT JOIN vehicle_data_source ds JOIN vehicle_data_source ds
ON ds.protocol = s2.protocol AND ds.source_ip = s2.source_ip ON ds.protocol = s2.protocol AND ds.source_ip = s2.source_ip
WHERE s2.vin = ? WHERE s2.vin = ?
AND s2.stat_date = ? AND s2.stat_date = ?
AND s2.protocol = ? AND s2.protocol = ?
AND s2.quality_status = '` + QualityOK + `' AND s2.quality_status = '` + QualityOK + `'
AND s2.daily_mileage_km BETWEEN 0 AND ? AND s2.daily_mileage_km BETWEEN 0 AND ?
AND COALESCE(ds.enabled, 1) = 1 AND ds.enabled = 1
ORDER BY COALESCE(ds.trust_priority, 100), ORDER BY ds.trust_priority,
s2.sample_count DESC, s2.sample_count DESC,
s2.latest_event_time DESC, s2.latest_event_time DESC,
s2.source_key ASC s2.source_key ASC

26
go/vehicle-gateway/internal/stats/source_mileage_test.go
View File

@@ -101,21 +101,27 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) {
for _, want := range []string{ for _, want := range []string{
"INSERT INTO vehicle_daily_mileage", "INSERT INTO vehicle_daily_mileage",
"FROM vehicle_daily_mileage_source s", "FROM vehicle_daily_mileage_source s",
"LEFT JOIN vehicle_data_source ds", "JOIN vehicle_data_source ds",
"ORDER BY COALESCE(ds.trust_priority, 100)", "ds.id",
"ORDER BY ds.trust_priority",
"s.quality_status = '" + QualityOK + "'", "s.quality_status = '" + QualityOK + "'",
"s.daily_mileage_km BETWEEN 0 AND", "s.daily_mileage_km BETWEEN 0 AND",
"ds.latest_source_endpoint",
} { } {
if !strings.Contains(projectFinal, want) { if !strings.Contains(projectFinal, want) {
t.Fatalf("project query missing %q: %s", want, projectFinal) t.Fatalf("project query missing %q: %s", want, projectFinal)
} }
} }
if strings.Contains(projectFinal, "COALESCE(NULLIF(ds.latest_source_endpoint, ''), s.source_endpoint)") { for _, forbidden := range []string{
t.Fatalf("project query should not fallback to source endpoint: %s", projectFinal) "trusted_source_key",
} "trusted_phone",
if strings.Contains(projectFinal, ", s.source_endpoint") { "trusted_source_endpoint",
t.Fatalf("project query should only project latest_source_endpoint: %s", projectFinal) "first_total_mileage_km",
"sample_count)",
"ds.latest_source_endpoint",
} {
if strings.Contains(projectFinal, forbidden) {
t.Fatalf("project query should not use final-table field %q: %s", forbidden, projectFinal)
}
} }
if !strings.Contains(markSelected, "UPDATE vehicle_daily_mileage_source s") || !strings.Contains(markSelected, "SET s.is_selected = 1") { if !strings.Contains(markSelected, "UPDATE vehicle_daily_mileage_source s") || !strings.Contains(markSelected, "SET s.is_selected = 1") {
t.Fatalf("mark query should flag the elected source: %s", markSelected) t.Fatalf("mark query should flag the elected source: %s", markSelected)
@@ -129,8 +135,8 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) {
"s2.source_key", "s2.source_key",
"s2.quality_status = '" + QualityOK + "'", "s2.quality_status = '" + QualityOK + "'",
"s2.daily_mileage_km BETWEEN 0 AND", "s2.daily_mileage_km BETWEEN 0 AND",
"COALESCE(ds.enabled, 1) = 1", "ds.enabled = 1",
"ORDER BY COALESCE(ds.trust_priority, 100)", "ORDER BY ds.trust_priority",
"LIMIT 1", "LIMIT 1",
} { } {
if !strings.Contains(markSelected, want) { if !strings.Contains(markSelected, want) {