diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index 5e37e461..9f9b2140 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -18,12 +18,14 @@ type Execer interface { } type Writer struct { - exec Execer - query Queryer - loc *time.Location - mu sync.Mutex - lastTotalMileage map[string]float64 - baselineCache map[string]sourceBaselineCacheEntry + exec Execer + query Queryer + loc *time.Location + sourceTouchInterval time.Duration + mu sync.Mutex + lastTotalMileage map[string]float64 + lastSourceSeen map[string]time.Time + baselineCache map[string]sourceBaselineCacheEntry } type MetricSample struct { @@ -46,10 +48,12 @@ func NewWriter(exec Execer, loc *time.Location) *Writer { loc = time.FixedZone("Asia/Shanghai", 8*3600) } writer := &Writer{ - exec: exec, - loc: loc, - lastTotalMileage: map[string]float64{}, - baselineCache: map[string]sourceBaselineCacheEntry{}, + exec: exec, + loc: loc, + sourceTouchInterval: time.Minute, + lastTotalMileage: map[string]float64{}, + lastSourceSeen: map[string]time.Time{}, + baselineCache: map[string]sourceBaselineCacheEntry{}, } if query, ok := exec.(Queryer); ok { writer.query = query @@ -76,11 +80,20 @@ func (w *Writer) EnsureSchema(ctx context.Context) error { } func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { + identity, hasSource := NewSourceIdentity(env.Protocol, env.SourceEndpoint) + if hasSource { + seenAt := w.sourceSeenAt(env) + if w.shouldTouchSource(identity, seenAt) { + if err := UpsertDataSource(ctx, w.exec, identity, seenAt); err != nil { + return err + } + w.markSourceTouched(identity, seenAt) + } + } samples, err := SamplesFromEnvelope(env, w.loc) if err != nil { return err } - identity, hasSource := NewSourceIdentity(env.Protocol, env.SourceEndpoint) for _, sample := range samples { if w.seenSameMileage(sample) { continue @@ -88,9 +101,6 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { if !hasSource { continue } - if err := UpsertDataSource(ctx, w.exec, identity, sample.EventTime); err != nil { - return err - } candidate := SourceMileageSampleFromMetric(sample, identity) if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil { return err @@ -106,6 +116,35 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { return nil } +func (w *Writer) sourceSeenAt(env envelope.FrameEnvelope) time.Time { + eventMS := env.EventTimeMS + if eventMS <= 0 { + eventMS = env.ReceivedAtMS + } + if eventMS > 0 { + return time.UnixMilli(eventMS).In(w.loc) + } + return time.Now().In(w.loc) +} + +func (w *Writer) shouldTouchSource(identity SourceIdentity, seenAt time.Time) bool { + key := string(identity.Protocol) + "|" + identity.SourceIP + w.mu.Lock() + defer w.mu.Unlock() + last, ok := w.lastSourceSeen[key] + if ok && !seenAt.After(last.Add(w.sourceTouchInterval)) { + return false + } + return true +} + +func (w *Writer) markSourceTouched(identity SourceIdentity, seenAt time.Time) { + key := string(identity.Protocol) + "|" + identity.SourceIP + w.mu.Lock() + w.lastSourceSeen[key] = seenAt + w.mu.Unlock() +} + func (w *Writer) seenSameMileage(sample MetricSample) bool { key := mileageCacheKey(sample) w.mu.Lock() diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index b9725f39..21fa936f 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -211,6 +211,33 @@ func TestWriterAppendSkipsMileageWithoutSourceIdentity(t *testing.T) { } } +func TestWriterAppendTracksSourceWithoutMileageSample(t *testing.T) { + exec := &recordingExec{} + writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) + event := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + Phone: "13307795518", + SourceEndpoint: "222.66.200.68:43614", + EventTimeMS: time.Date(2026, 7, 8, 16, 30, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), + Fields: map[string]any{ + "jt808.location.speed_kmh": 38.5, + }, + } + + if err := writer.Append(context.Background(), event); err != nil { + t.Fatalf("Append() error = %v", err) + } + if len(exec.calls) != 1 { + t.Fatalf("exec calls = %d, want source only", len(exec.calls)) + } + if !strings.Contains(exec.calls[0].query, "INSERT INTO vehicle_data_source") { + t.Fatalf("source upsert query missing: %s", exec.calls[0].query) + } + if got := exec.calls[0].args[1]; got != "222.66.200.68" { + t.Fatalf("source ip arg = %#v", got) + } +} + func TestWriterAppendDedupesRealtimeMileagePerSource(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))