From 762c7265d770e8a3952af9f9c9e1c3b3e77d5d3f Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 8 Jul 2026 14:48:51 +0800 Subject: [PATCH] feat(stats): write realtime mileage candidates --- .superpowers/sdd/task-4-report.md | 6 + .../internal/stats/daily_metric.go | 106 ++---------------- .../internal/stats/daily_metric_test.go | 83 ++++++++++---- 3 files changed, 80 insertions(+), 115 deletions(-) create mode 100644 .superpowers/sdd/task-4-report.md diff --git a/.superpowers/sdd/task-4-report.md b/.superpowers/sdd/task-4-report.md new file mode 100644 index 00000000..6077177d --- /dev/null +++ b/.superpowers/sdd/task-4-report.md @@ -0,0 +1,6 @@ +Status: Completed Task 4. +Commit: 821ea8e39 (`feat(stats): write realtime mileage candidates`). +Test Summary: `go test ./internal/stats -run TestWriterAppendWritesSourceCandidateAndProjection -count=1` (PASS) and `go test ./internal/stats -count=1` (PASS). +Concerns: `Writer.Append` now always runs projector projection per sample event, even when source identity is unavailable; this preserves final row refresh behavior and matches existing schema flow. +Files changed: `go/vehicle-gateway/internal/stats/daily_metric.go`, `go/vehicle-gateway/internal/stats/daily_metric_test.go`. +Report path: `.superpowers/sdd/task-4-report.md`. diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index 32c809f1..dcef0def 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -69,19 +69,21 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { if err != nil { return err } + identity, hasSource := NewSourceIdentity(env.Protocol, env.SourceEndpoint) for _, sample := range samples { if w.seenSameMileage(sample) { continue } - if _, err := w.exec.ExecContext(ctx, upsertDailyMileageSQL, - sample.VIN, - sample.StatDate, - string(sample.Protocol), - sample.TotalMileageKM, - sample.TotalMileageKM, - sample.SourceKey, - sample.Phone, - sample.SourceEndpoint); err != nil { + if hasSource { + if err := UpsertDataSource(ctx, w.exec, identity, sample.EventTime); err != nil { + return err + } + candidate := SourceMileageSampleFromMetric(sample, identity) + if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil { + return err + } + } + if err := ProjectDailyMileage(ctx, w.exec, sample.VIN, sample.StatDate, sample.Protocol); err != nil { return err } w.markMileageWritten(sample) @@ -196,92 +198,6 @@ func mileageMappingsByProtocol(protocol envelope.Protocol) []mileageFieldMapping } } -const upsertDailyMileageSQL = ` -INSERT INTO vehicle_daily_mileage - (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) -VALUES (?, ?, ?, 0, ?, ?, ?, ?, ?, 1) -ON DUPLICATE KEY UPDATE - first_total_mileage_km = CASE - WHEN trusted_source_key IS NOT NULL - AND trusted_source_key <> '' - AND trusted_source_key <> VALUES(trusted_source_key) - AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 - THEN first_total_mileage_km - 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 trusted_source_key IS NOT NULL - AND trusted_source_key <> '' - AND trusted_source_key <> VALUES(trusted_source_key) - AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 - THEN latest_total_mileage_km - 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 trusted_source_key IS NOT NULL - AND trusted_source_key <> '' - AND trusted_source_key <> VALUES(trusted_source_key) - AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 - THEN latest_total_mileage_km - 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, - CASE - WHEN trusted_source_key IS NOT NULL - AND trusted_source_key <> '' - AND trusted_source_key <> VALUES(trusted_source_key) - AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 - THEN latest_total_mileage_km - ELSE VALUES(latest_total_mileage_km) - END - ) - CASE - WHEN trusted_source_key IS NOT NULL - AND trusted_source_key <> '' - AND trusted_source_key <> VALUES(trusted_source_key) - AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 - THEN first_total_mileage_km - 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, - trusted_source_key = CASE - WHEN trusted_source_key IS NULL OR trusted_source_key = '' - THEN VALUES(trusted_source_key) - WHEN trusted_source_key = VALUES(trusted_source_key) - THEN trusted_source_key - WHEN ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) <= 50 - THEN VALUES(trusted_source_key) - ELSE trusted_source_key - END, - trusted_phone = CASE - WHEN trusted_source_key IS NULL OR trusted_source_key = '' OR trusted_source_key = VALUES(trusted_source_key) - THEN VALUES(trusted_phone) - ELSE trusted_phone - END, - trusted_source_endpoint = CASE - WHEN trusted_source_key IS NULL OR trusted_source_key = '' OR trusted_source_key = VALUES(trusted_source_key) - THEN VALUES(trusted_source_endpoint) - ELSE trusted_source_endpoint - END, - sample_count = CASE - WHEN trusted_source_key IS NOT NULL - AND trusted_source_key <> '' - AND trusted_source_key <> VALUES(trusted_source_key) - AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 - THEN sample_count - ELSE sample_count + 1 - END, - updated_at = CURRENT_TIMESTAMP -` - func floatField(env envelope.FrameEnvelope, key string) (float64, bool) { if env.Fields == nil { return 0, false diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 0cb9f905..efd9c070 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -156,6 +156,37 @@ func TestSamplesFromEnvelopeSkipsNonPositiveMileage(t *testing.T) { } } +func TestWriterAppendWritesSourceCandidateAndProjection(t *testing.T) { + exec := &recordingExec{} + writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) + event := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "LA9GG64L7PBAF4001", + Phone: "13307765812", + SourceEndpoint: "115.231.168.135:20215", + EventTimeMS: time.Date(2026, 7, 8, 13, 20, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), + Fields: map[string]any{ + "jt808.location.total_mileage_km": 4123.9, + }, + } + + if err := writer.Append(context.Background(), event); err != nil { + t.Fatalf("Append() error = %v", err) + } + if len(exec.calls) != 5 { + t.Fatalf("exec calls = %d, want source + candidate + clear + project + mark", len(exec.calls)) + } + if !strings.Contains(exec.calls[0].query, "INSERT INTO vehicle_data_source") { + t.Fatalf("first call should upsert source: %s", exec.calls[0].query) + } + if !strings.Contains(exec.calls[1].query, "INSERT INTO vehicle_daily_mileage_source") { + t.Fatalf("second call should upsert candidate: %s", exec.calls[1].query) + } + if !strings.Contains(exec.calls[3].query, "INSERT INTO vehicle_daily_mileage") { + t.Fatalf("fourth call should project final: %s", exec.calls[3].query) + } +} + func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) @@ -163,9 +194,11 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { t.Fatalf("EnsureSchema() error = %v", err) } if err := writer.Append(context.Background(), envelope.FrameEnvelope{ - Protocol: envelope.ProtocolGB32960, - VIN: "LNBVIN00000000002", - EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), + Protocol: envelope.ProtocolGB32960, + VIN: "LNBVIN00000000002", + Phone: "13307765812", + SourceEndpoint: "115.231.168.135:20215", + EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), Fields: map[string]any{ envelope.FieldTotalMileageKM: "10000.0", }, @@ -173,7 +206,7 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { t.Fatalf("Append() error = %v", err) } - if len(exec.calls) != 7 { + if len(exec.calls) != 11 { t.Fatalf("exec calls = %d", len(exec.calls)) } for i, want := range []string{ @@ -196,22 +229,32 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { 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) + projectCall := exec.calls[9] + if !strings.Contains(projectCall.query, "INSERT INTO vehicle_daily_mileage") { + t.Fatalf("unexpected project sql: %s", projectCall.query) } - if strings.Contains(upsertCall.query, "metric_key") || strings.Contains(upsertCall.query, "metric_unit") { - t.Fatalf("daily mileage upsert should not use generic metric columns: %s", upsertCall.query) + if !strings.Contains(projectCall.query, "ON DUPLICATE KEY UPDATE") { + t.Fatalf("unexpected upsert sql: %s", projectCall.query) } - if !strings.Contains(upsertCall.query, "first_total_mileage_km <= 0") { - t.Fatalf("upsert should ignore legacy zero first mileage: %s", upsertCall.query) + 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) } - if !strings.Contains(upsertCall.query, "trusted_source_key") { - t.Fatalf("upsert should track trusted source: %s", upsertCall.query) + if !strings.Contains(projectCall.query, "LEFT JOIN vehicle_data_source ds") { + t.Fatalf("project sql should join data source: %s", projectCall.query) } - if got := upsertCall.args[0]; got != "LNBVIN00000000002" { + if !strings.Contains(projectCall.query, "trusted_source_key") { + t.Fatalf("upsert should track trusted source: %s", projectCall.query) + } + if got := projectCall.args[0]; got != "LNBVIN00000000002" { t.Fatalf("first upsert arg should be vin, got %#v", got) } + + if !strings.Contains(exec.calls[6].query, "INSERT INTO vehicle_data_source") { + t.Fatalf("source upsert query missing: %s", exec.calls[6].query) + } + if !strings.Contains(exec.calls[7].query, "INSERT INTO vehicle_daily_mileage_source") { + t.Fatalf("candidate upsert query missing: %s", exec.calls[7].query) + } } func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) { @@ -234,8 +277,8 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) { t.Fatalf("duplicate Append() error = %v", err) } - if len(exec.calls) != 1 { - t.Fatalf("exec calls = %d, want 1", len(exec.calls)) + if len(exec.calls) != 3 { + t.Fatalf("exec calls = %d, want 3", len(exec.calls)) } } @@ -258,8 +301,8 @@ func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) { t.Fatalf("retry Append() error = %v", err) } - if len(exec.calls) != 2 { - t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + if len(exec.calls) != 4 { + t.Fatalf("exec calls = %d, want 4", len(exec.calls)) } } @@ -283,8 +326,8 @@ func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) { t.Fatalf("next-day Append() error = %v", err) } - if len(exec.calls) != 2 { - t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + if len(exec.calls) != 6 { + t.Fatalf("exec calls = %d, want 6", len(exec.calls)) } if len(writer.lastTotalMileage) != 1 { t.Fatalf("cache entries = %d, want 1", len(writer.lastTotalMileage))