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

Fix candidate mileage review findings

Browse Source
This commit is contained in:
lingniu
2026-07-08 14:29:54 +08:00
parent 08e2d8c0f0
commit b7f6e47ce3
5 changed files with 82 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -47,8 +47,14 @@ func NewWriter(exec Execer, loc *time.Location) *Writer {
}
func (w *Writer) EnsureSchema(ctx context.Context) error {
if _, err := w.exec.ExecContext(ctx, DailyMileageTableSQL); err != nil {
return err
for _, statement := range []string{
DataSourceTableSQL,
DailyMileageSourceTableSQL,
DailyMileageTableSQL,
} {
if _, err := w.exec.ExecContext(ctx, statement); err != nil {
return err
}
}
for _, statement := range DailyMileageAlterSQL {
if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) {

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

@@ -173,24 +173,30 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
t.Fatalf("Append() error = %v", err)
}
if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_daily_mileage") {
t.Fatalf("unexpected schema sql: %s", exec.calls[0].query)
}
for _, column := range []string{"vehicle_key", "id BIGINT", "AUTO_INCREMENT", "created_at"} {
if strings.Contains(exec.calls[0].query, column) {
t.Fatalf("schema should not include %s: %s", column, exec.calls[0].query)
}
}
if !strings.Contains(exec.calls[0].query, "PRIMARY KEY (vin, stat_date, protocol)") {
t.Fatalf("daily mileage table should key by vin/stat_date/protocol: %s", exec.calls[0].query)
}
if strings.Contains(exec.calls[0].query, "KEY idx_vin (vin)") {
t.Fatalf("daily mileage table should not keep redundant vin index covered by the primary key: %s", exec.calls[0].query)
}
if len(exec.calls) != 5 {
if len(exec.calls) != 7 {
t.Fatalf("exec calls = %d", len(exec.calls))
}
upsertCall := exec.calls[4]
for i, want := range []string{
"CREATE TABLE IF NOT EXISTS vehicle_data_source",
"CREATE TABLE IF NOT EXISTS vehicle_daily_mileage_source",
"CREATE TABLE IF NOT EXISTS vehicle_daily_mileage",
} {
if !strings.Contains(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"} {
if strings.Contains(exec.calls[2].query, column) {
t.Fatalf("schema should not include %s: %s", column, exec.calls[2].query)
}
}
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)
}
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)
}
upsertCall := exec.calls[6]
if !strings.Contains(upsertCall.query, "ON DUPLICATE KEY UPDATE") {
t.Fatalf("unexpected upsert sql: %s", upsertCall.query)
}

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

@@ -71,7 +71,7 @@ func UpsertSourceMileage(ctx context.Context, exec Execer, sample SourceMileageS
if exec == nil {
panic("stats execer must not be nil")
}
if sample.VIN == "" || sample.SourceKey == "" {
if sample.VIN == "" || sample.SourceKey == "" || strings.TrimSpace(sample.SourceIP) == "" {
return nil
}
if sample.QualityStatus == "" {
@@ -105,27 +105,26 @@ INSERT INTO vehicle_daily_mileage_source
first_total_mileage_km, latest_total_mileage_km, daily_mileage_km, sample_count,
first_event_time, latest_event_time, quality_status, quality_reason)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
source_ip = VALUES(source_ip),
source_endpoint = VALUES(source_endpoint),
phone = VALUES(phone),
device_id = VALUES(device_id),
platform_name = COALESCE(VALUES(platform_name), platform_name),
first_total_mileage_km = CASE
WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
THEN VALUES(first_total_mileage_km)
ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
END,
latest_total_mileage_km = CASE
WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
THEN VALUES(latest_total_mileage_km)
ELSE GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
END,
/* daily_mileage_km = VALUES(daily_mileage_km) */
daily_mileage_km = GREATEST(
CASE
WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
THEN VALUES(latest_total_mileage_km)
ON DUPLICATE KEY UPDATE
source_ip = VALUES(source_ip),
source_endpoint = VALUES(source_endpoint),
phone = VALUES(phone),
device_id = VALUES(device_id),
platform_name = COALESCE(NULLIF(TRIM(VALUES(platform_name)), ''), platform_name),
first_total_mileage_km = CASE
WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
THEN VALUES(first_total_mileage_km)
ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
END,
latest_total_mileage_km = CASE
WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
THEN VALUES(latest_total_mileage_km)
ELSE GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
END,
daily_mileage_km = GREATEST(
CASE
WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
THEN VALUES(latest_total_mileage_km)
ELSE latest_total_mileage_km
END,
VALUES(latest_total_mileage_km)

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

@@ -50,8 +50,9 @@ func TestUpsertSourceMileageWritesCandidateRow(t *testing.T) {
for _, want := range []string{
"INSERT INTO vehicle_daily_mileage_source",
"ON DUPLICATE KEY UPDATE",
"daily_mileage_km = VALUES(daily_mileage_km)",
"daily_mileage_km = GREATEST(",
"quality_status = VALUES(quality_status)",
"platform_name = COALESCE(NULLIF(TRIM(VALUES(platform_name)), ''), platform_name)",
} {
if !strings.Contains(sql, want) {
t.Fatalf("candidate upsert missing %q: %s", want, sql)
@@ -61,3 +62,22 @@ func TestUpsertSourceMileageWritesCandidateRow(t *testing.T) {
t.Fatalf("first arg = %#v", got)
}
}
func TestUpsertSourceMileageSkipsBlankSourceIP(t *testing.T) {
exec := &recordingExec{}
sample := SourceMileageSample{
VIN: "LA9GG64L7PBAF4001",
StatDate: "2026-07-08",
Protocol: envelope.ProtocolJT808,
SourceKey: "JT808:13307765812@",
SourceIP: " ",
Phone: "13307765812",
QualityStatus: QualityOK,
}
if err := UpsertSourceMileage(context.Background(), exec, sample); err != nil {
t.Fatalf("UpsertSourceMileage() error = %v", err)
}
if len(exec.calls) != 0 {
t.Fatalf("exec calls = %d, want 0", len(exec.calls))
}
}