shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
go
lingniu-vehicle-ingest/docs/superpowers/plans/2026-07-08-multi-source-mileage-stats-implementation.md

40 KiB
Raw Permalink Blame History

Multi-Source Mileage Statistics Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Store every mileage source candidate, maintain source platform metadata, and project one elected daily mileage result for default BI/API queries.

Architecture: Add a source metadata table and a per-source daily mileage candidate table in MySQL. Real-time stat writer and historical backfill write candidates first, then run an election projector that updates the existing vehicle_daily_mileage final table. Default reads stay on vehicle_daily_mileage; source audit reads can be added after the write path is stable.

Tech Stack: Go 1.26, MySQL via github.com/go-sql-driver/mysql, TDengine via github.com/taosdata/driver-go/v3/taosWS, Kafka via github.com/segmentio/kafka-go, existing internal/stats package.

Global Constraints

  • Runtime flow must not add non-protocol synthetic fields to vehicle.fields.go.*.
  • Source identity uses protocol + source_ip; source endpoint port is latest connection metadata only.
  • Platform name, trust priority, enabled flag, and remark are operator-maintained and must not be overwritten by runtime upserts.
  • Candidate statistics use vin + stat_date + protocol + source_key and never mix sources.
  • Final statistics use vin + stat_date + protocol and are projected from one selected candidate.
  • The projector must not calculate B today - A yesterday.
  • A source with no same-source previous-day baseline remains a candidate and is not selected by default.
  • Default API queries continue to read vehicle_daily_mileage.

File Structure

  • Modify go/vehicle-gateway/internal/stats/schema.go
    • Owns MySQL schema strings for source metadata, source candidates, final table, and compatibility alters.
  • Create go/vehicle-gateway/internal/stats/source.go
    • Owns source endpoint normalization, SourceIdentity, and source metadata upsert.
  • Create go/vehicle-gateway/internal/stats/source_test.go
    • Tests source IP extraction and runtime upsert preserving manual fields.
  • Create go/vehicle-gateway/internal/stats/source_mileage.go
    • Owns SourceMileageSample, candidate upsert, final election projection, and shared constants.
  • Create go/vehicle-gateway/internal/stats/source_mileage_test.go
    • Tests candidate writes and election rules.
  • Modify go/vehicle-gateway/internal/stats/daily_metric.go
    • Keep protocol mileage extraction, route Writer.Append through source metadata, candidate upsert, and election projection.
  • Modify go/vehicle-gateway/internal/stats/daily_metric_test.go
    • Update expectations for source candidate writes and final projection.
  • Modify go/vehicle-gateway/internal/stats/query.go
    • Keep default response compatible; optionally include selected source columns in existing rows if already present.
  • Modify go/vehicle-gateway/cmd/stats-backfill/main.go
    • Rebuild source candidates by day/source, then project final rows.
  • Modify go/vehicle-gateway/cmd/stats-backfill/main_test.go
    • Update trusted-source tests to assert source candidates and final selection.
  • Deploy binaries:
    • /opt/lingniu-go-native/current/stat-writer
    • /opt/lingniu-go-native/current/stats-backfill

Task 1: Source Metadata Schema And Identity Helpers

Files:

  • Modify: go/vehicle-gateway/internal/stats/schema.go
  • Create: go/vehicle-gateway/internal/stats/source.go
  • Create: go/vehicle-gateway/internal/stats/source_test.go

Interfaces:

  • Produces: type SourceIdentity struct { Protocol envelope.Protocol; SourceIP string; SourceEndpoint string }

  • Produces: func NewSourceIdentity(protocol envelope.Protocol, endpoint string) (SourceIdentity, bool)

  • Produces: func NormalizeSourceIP(endpoint string) string

  • Produces: func UpsertDataSource(ctx context.Context, exec Execer, identity SourceIdentity, now time.Time) error

  • Consumes: stats.Execer

  • Step 1: Write failing source identity tests

Add to go/vehicle-gateway/internal/stats/source_test.go:

package stats

import (
	"context"
	"strings"
	"testing"
	"time"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)

func TestNormalizeSourceIPDropsPort(t *testing.T) {
	tests := map[string]string{
		"115.231.168.135:20215": "115.231.168.135",
		"115.231.168.135":       "115.231.168.135",
		" 115.159.85.149:28316 ": "115.159.85.149",
		"":                      "",
	}
	for input, want := range tests {
		if got := NormalizeSourceIP(input); got != want {
			t.Fatalf("NormalizeSourceIP(%q) = %q, want %q", input, got, want)
		}
	}
}

func TestNewSourceIdentityRequiresSourceIP(t *testing.T) {
	identity, ok := NewSourceIdentity(envelope.ProtocolJT808, "115.231.168.135:20215")
	if !ok {
		t.Fatal("NewSourceIdentity() ok = false")
	}
	if identity.Protocol != envelope.ProtocolJT808 {
		t.Fatalf("protocol = %q", identity.Protocol)
	}
	if identity.SourceIP != "115.231.168.135" {
		t.Fatalf("source ip = %q", identity.SourceIP)
	}
	if identity.SourceEndpoint != "115.231.168.135:20215" {
		t.Fatalf("endpoint = %q", identity.SourceEndpoint)
	}

	if _, ok := NewSourceIdentity(envelope.ProtocolJT808, ""); ok {
		t.Fatal("empty endpoint should not produce identity")
	}
}

func TestUpsertDataSourcePreservesManualFields(t *testing.T) {
	exec := &recordingExec{}
	identity := SourceIdentity{
		Protocol:       envelope.ProtocolJT808,
		SourceIP:       "115.231.168.135",
		SourceEndpoint: "115.231.168.135:20215",
	}
	if err := UpsertDataSource(context.Background(), exec, identity, time.Date(2026, 7, 8, 13, 0, 0, 0, time.UTC)); err != nil {
		t.Fatalf("UpsertDataSource() error = %v", err)
	}
	if len(exec.calls) != 1 {
		t.Fatalf("exec calls = %d", len(exec.calls))
	}
	sql := exec.calls[0].query
	for _, want := range []string{
		"INSERT INTO vehicle_data_source",
		"latest_source_endpoint = VALUES(latest_source_endpoint)",
		"latest_seen_at = VALUES(latest_seen_at)",
	} {
		if !strings.Contains(sql, want) {
			t.Fatalf("source upsert missing %q: %s", want, sql)
		}
	}
	for _, forbidden := range []string{
		"platform_name = VALUES(platform_name)",
		"trust_priority = VALUES(trust_priority)",
		"enabled = VALUES(enabled)",
		"remark = VALUES(remark)",
	} {
		if strings.Contains(sql, forbidden) {
			t.Fatalf("source upsert should preserve manual field %q: %s", forbidden, sql)
		}
	}
}
  • Step 2: Run source tests and verify failure

Run:

cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
go test ./internal/stats -run 'TestNormalizeSourceIP|TestNewSourceIdentity|TestUpsertDataSource' -count=1

Expected: FAIL because NormalizeSourceIP, NewSourceIdentity, and UpsertDataSource are undefined.

  • Step 3: Add source schema

Add to go/vehicle-gateway/internal/stats/schema.go:

const DataSourceTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_data_source (
  id BIGINT NOT NULL AUTO_INCREMENT,
  protocol VARCHAR(32) NOT NULL,
  source_ip VARCHAR(64) NOT NULL,
  latest_source_endpoint VARCHAR(128) NULL,
  platform_name VARCHAR(128) NULL,
  trust_priority INT NOT NULL DEFAULT 100,
  enabled TINYINT(1) NOT NULL DEFAULT 1,
  first_seen_at DATETIME NULL,
  latest_seen_at DATETIME NULL,
  remark VARCHAR(512) NULL,
  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (id),
  UNIQUE KEY uk_protocol_source_ip (protocol, source_ip),
  KEY idx_protocol_enabled_priority (protocol, enabled, trust_priority)
)`

Update Writer.EnsureSchema in Task 3 to execute this statement.

  • Step 4: Implement source helpers

Create go/vehicle-gateway/internal/stats/source.go:

package stats

import (
	"context"
	"strings"
	"time"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)

type SourceIdentity struct {
	Protocol       envelope.Protocol
	SourceIP       string
	SourceEndpoint string
}

func NewSourceIdentity(protocol envelope.Protocol, endpoint string) (SourceIdentity, bool) {
	sourceIP := NormalizeSourceIP(endpoint)
	if sourceIP == "" {
		return SourceIdentity{}, false
	}
	return SourceIdentity{
		Protocol:       protocol,
		SourceIP:       sourceIP,
		SourceEndpoint: strings.TrimSpace(endpoint),
	}, true
}

func NormalizeSourceIP(endpoint string) string {
	endpoint = strings.TrimSpace(endpoint)
	if endpoint == "" {
		return ""
	}
	if host, _, ok := strings.Cut(endpoint, ":"); ok {
		return strings.TrimSpace(host)
	}
	return endpoint
}

func UpsertDataSource(ctx context.Context, exec Execer, identity SourceIdentity, now time.Time) error {
	if exec == nil {
		panic("stats execer must not be nil")
	}
	if identity.SourceIP == "" {
		return nil
	}
	if now.IsZero() {
		now = time.Now()
	}
	_, err := exec.ExecContext(ctx, upsertDataSourceSQL,
		string(identity.Protocol),
		identity.SourceIP,
		identity.SourceEndpoint,
		now,
		now,
	)
	return err
}

const upsertDataSourceSQL = `
INSERT INTO vehicle_data_source
  (protocol, source_ip, latest_source_endpoint, first_seen_at, latest_seen_at)
VALUES (?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
  latest_source_endpoint = VALUES(latest_source_endpoint),
  latest_seen_at = VALUES(latest_seen_at),
  updated_at = CURRENT_TIMESTAMP
`
  • Step 5: Run source tests and verify pass

Run:

go test ./internal/stats -run 'TestNormalizeSourceIP|TestNewSourceIdentity|TestUpsertDataSource' -count=1

Expected: PASS.

  • Step 6: Commit source metadata foundation

Run:

git add go/vehicle-gateway/internal/stats/schema.go go/vehicle-gateway/internal/stats/source.go go/vehicle-gateway/internal/stats/source_test.go
git commit -m "feat(stats): add vehicle data source metadata"

Task 2: Candidate Mileage Schema And Writer

Files:

  • Modify: go/vehicle-gateway/internal/stats/schema.go
  • Create: go/vehicle-gateway/internal/stats/source_mileage.go
  • Create: go/vehicle-gateway/internal/stats/source_mileage_test.go

Interfaces:

  • Consumes: MetricSample

  • Produces: type SourceMileageSample struct

  • Produces: func SourceMileageSampleFromMetric(sample MetricSample, identity SourceIdentity) SourceMileageSample

  • Produces: func UpsertSourceMileage(ctx context.Context, exec Execer, sample SourceMileageSample) error

  • Produces: func SourceKey(protocol envelope.Protocol, phone string, deviceID string, sourceIP string) string

  • Step 1: Write failing candidate tests

Add to go/vehicle-gateway/internal/stats/source_mileage_test.go:

package stats

import (
	"context"
	"strings"
	"testing"
	"time"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)

func TestSourceKeyUsesProtocolDeviceAndSourceIP(t *testing.T) {
	got := SourceKey(envelope.ProtocolJT808, "13307765812", "", "115.231.168.135")
	if got != "JT808:13307765812@115.231.168.135" {
		t.Fatalf("source key = %q", got)
	}

	got = SourceKey(envelope.ProtocolGB32960, "", "iccid-1", "202.98.117.132")
	if got != "GB32960:iccid-1@202.98.117.132" {
		t.Fatalf("source key = %q", got)
	}
}

func TestUpsertSourceMileageWritesCandidateRow(t *testing.T) {
	exec := &recordingExec{}
	sample := SourceMileageSample{
		VIN:            "LA9GG64L7PBAF4001",
		StatDate:       "2026-07-08",
		Protocol:       envelope.ProtocolJT808,
		SourceKey:      "JT808:13307765812@115.231.168.135",
		SourceIP:       "115.231.168.135",
		SourceEndpoint: "115.231.168.135:20215",
		Phone:          "13307765812",
		FirstTotalKM:   4100.8,
		LatestTotalKM:  4123.9,
		DailyKM:        23.1,
		SampleCount:    2,
		FirstEventTime: time.Date(2026, 7, 8, 0, 1, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
		LatestEventTime: time.Date(2026, 7, 8, 23, 59, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
		QualityStatus:  QualityOK,
		QualityReason:  "same_source_continuous",
	}
	if err := UpsertSourceMileage(context.Background(), exec, sample); err != nil {
		t.Fatalf("UpsertSourceMileage() error = %v", err)
	}
	if len(exec.calls) != 1 {
		t.Fatalf("exec calls = %d", len(exec.calls))
	}
	sql := exec.calls[0].query
	for _, want := range []string{
		"INSERT INTO vehicle_daily_mileage_source",
		"ON DUPLICATE KEY UPDATE",
		"daily_mileage_km = VALUES(daily_mileage_km)",
		"quality_status = VALUES(quality_status)",
	} {
		if !strings.Contains(sql, want) {
			t.Fatalf("candidate upsert missing %q: %s", want, sql)
		}
	}
	if got := exec.calls[0].args[0]; got != "LA9GG64L7PBAF4001" {
		t.Fatalf("first arg = %#v", got)
	}
}
  • Step 2: Run candidate tests and verify failure

Run:

go test ./internal/stats -run 'TestSourceKey|TestUpsertSourceMileage' -count=1

Expected: FAIL because candidate interfaces are undefined.

  • Step 3: Add candidate schema

Add to go/vehicle-gateway/internal/stats/schema.go:

const DailyMileageSourceTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_mileage_source (
  vin VARCHAR(32) NOT NULL,
  stat_date DATE NOT NULL,
  protocol VARCHAR(32) NOT NULL,
  source_key VARCHAR(256) NOT NULL,
  source_ip VARCHAR(64) NOT NULL,
  source_endpoint VARCHAR(128) NULL,
  phone VARCHAR(32) NULL,
  device_id VARCHAR(64) NULL,
  platform_name VARCHAR(128) NULL,
  first_total_mileage_km DECIMAL(18,3) NULL,
  latest_total_mileage_km DECIMAL(18,3) NULL,
  daily_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0,
  sample_count BIGINT NOT NULL DEFAULT 0,
  first_event_time DATETIME NULL,
  latest_event_time DATETIME NULL,
  quality_status VARCHAR(32) NOT NULL DEFAULT 'OK',
  quality_reason VARCHAR(255) NULL,
  is_selected TINYINT(1) NOT NULL DEFAULT 0,
  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (vin, stat_date, protocol, source_key),
  KEY idx_protocol_date (protocol, stat_date),
  KEY idx_source_ip (protocol, source_ip),
  KEY idx_selected (stat_date, protocol, is_selected)
)`
  • Step 4: Implement candidate writer

Create go/vehicle-gateway/internal/stats/source_mileage.go:

package stats

import (
	"context"
	"strings"
	"time"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)

const (
	QualityOK                 = "OK"
	QualityNoPreviousBaseline = "NO_PREVIOUS_BASELINE"
	QualityInvalidDelta       = "INVALID_DELTA"
)

type SourceMileageSample struct {
	VIN             string
	StatDate        string
	Protocol        envelope.Protocol
	SourceKey       string
	SourceIP        string
	SourceEndpoint  string
	Phone           string
	DeviceID        string
	PlatformName    string
	FirstTotalKM    float64
	LatestTotalKM   float64
	DailyKM         float64
	SampleCount     int64
	FirstEventTime  time.Time
	LatestEventTime time.Time
	QualityStatus   string
	QualityReason   string
}

func SourceKey(protocol envelope.Protocol, phone string, deviceID string, sourceIP string) string {
	identity := strings.TrimSpace(phone)
	if identity == "" {
		identity = strings.TrimSpace(deviceID)
	}
	if identity == "" {
		identity = "unknown"
	}
	return string(protocol) + ":" + identity + "@" + strings.TrimSpace(sourceIP)
}

func SourceMileageSampleFromMetric(sample MetricSample, identity SourceIdentity) SourceMileageSample {
	eventTime := sample.EventTime
	return SourceMileageSample{
		VIN:             sample.VIN,
		StatDate:        sample.StatDate,
		Protocol:        sample.Protocol,
		SourceKey:       SourceKey(sample.Protocol, sample.Phone, sample.DeviceID, identity.SourceIP),
		SourceIP:        identity.SourceIP,
		SourceEndpoint:  identity.SourceEndpoint,
		Phone:           sample.Phone,
		DeviceID:        sample.DeviceID,
		FirstTotalKM:    sample.TotalMileageKM,
		LatestTotalKM:   sample.TotalMileageKM,
		DailyKM:         0,
		SampleCount:     1,
		FirstEventTime:  eventTime,
		LatestEventTime: eventTime,
		QualityStatus:   QualityOK,
		QualityReason:   "realtime_sample",
	}
}

func UpsertSourceMileage(ctx context.Context, exec Execer, sample SourceMileageSample) error {
	if sample.VIN == "" || sample.SourceKey == "" {
		return nil
	}
	if sample.QualityStatus == "" {
		sample.QualityStatus = QualityOK
	}
	_, err := exec.ExecContext(ctx, upsertSourceMileageSQL,
		sample.VIN,
		sample.StatDate,
		string(sample.Protocol),
		sample.SourceKey,
		sample.SourceIP,
		sample.SourceEndpoint,
		sample.Phone,
		sample.DeviceID,
		sample.PlatformName,
		sample.FirstTotalKM,
		sample.LatestTotalKM,
		sample.DailyKM,
		sample.SampleCount,
		sample.FirstEventTime,
		sample.LatestEventTime,
		sample.QualityStatus,
		sample.QualityReason,
	)
	return err
}

const upsertSourceMileageSQL = `
INSERT INTO vehicle_daily_mileage_source
  (vin, stat_date, protocol, source_key, source_ip, source_endpoint, phone, device_id, platform_name,
   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 = 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)
  ) - 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,
  sample_count = sample_count + VALUES(sample_count),
  first_event_time = CASE
    WHEN first_event_time IS NULL OR VALUES(first_event_time) < first_event_time
      THEN VALUES(first_event_time)
    ELSE first_event_time
  END,
  latest_event_time = CASE
    WHEN latest_event_time IS NULL OR VALUES(latest_event_time) > latest_event_time
      THEN VALUES(latest_event_time)
    ELSE latest_event_time
  END,
  quality_status = VALUES(quality_status),
  quality_reason = VALUES(quality_reason),
  updated_at = CURRENT_TIMESTAMP
`
  • Step 5: Extend MetricSample with event and device fields

Modify MetricSample in go/vehicle-gateway/internal/stats/daily_metric.go:

type MetricSample struct {
	VIN            string
	Protocol       envelope.Protocol
	StatDate       string
	TotalMileageKM float64
	EventTime       time.Time
	SourceKey      string
	Phone          string
	DeviceID       string
	SourceEndpoint string
}

Populate EventTime and DeviceID in SamplesFromEnvelope.

  • Step 6: Run candidate tests and verify pass

Run:

go test ./internal/stats -run 'TestSourceKey|TestUpsertSourceMileage' -count=1

Expected: PASS.

  • Step 7: Commit candidate writer

Run:

git add go/vehicle-gateway/internal/stats/schema.go go/vehicle-gateway/internal/stats/source_mileage.go go/vehicle-gateway/internal/stats/source_mileage_test.go go/vehicle-gateway/internal/stats/daily_metric.go
git commit -m "feat(stats): store source mileage candidates"

Task 3: Final Mileage Election Projector

Files:

  • Modify: go/vehicle-gateway/internal/stats/source_mileage.go
  • Modify: go/vehicle-gateway/internal/stats/source_mileage_test.go
  • Modify: go/vehicle-gateway/internal/stats/schema.go

Interfaces:

  • Produces: func ProjectDailyMileage(ctx context.Context, exec Execer, vin string, statDate string, protocol envelope.Protocol) error

  • Produces: const maxSelectedDailyMileageKM = 1000

  • Consumes: vehicle_daily_mileage_source

  • Produces: selected row in vehicle_daily_mileage

  • Step 1: Write failing projector SQL tests

Add to go/vehicle-gateway/internal/stats/source_mileage_test.go:

func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) {
	exec := &recordingExec{}
	err := ProjectDailyMileage(context.Background(), exec, "LA9GG64L7PBAF4001", "2026-07-08", envelope.ProtocolJT808)
	if err != nil {
		t.Fatalf("ProjectDailyMileage() error = %v", err)
	}
	if len(exec.calls) != 2 {
		t.Fatalf("exec calls = %d, want 2", len(exec.calls))
	}
	updateSources := exec.calls[0].query
	projectFinal := exec.calls[1].query
	if !strings.Contains(updateSources, "UPDATE vehicle_daily_mileage_source") || !strings.Contains(updateSources, "is_selected = 0") {
		t.Fatalf("first query should clear selected candidates: %s", updateSources)
	}
	for _, want := range []string{
		"INSERT INTO vehicle_daily_mileage",
		"FROM vehicle_daily_mileage_source s",
		"LEFT JOIN vehicle_data_source ds",
		"ORDER BY COALESCE(ds.trust_priority, 100)",
		"s.quality_status = 'OK'",
		"s.daily_mileage_km BETWEEN 0 AND",
	} {
		if !strings.Contains(projectFinal, want) {
			t.Fatalf("project query missing %q: %s", want, projectFinal)
		}
	}
}
  • Step 2: Run projector test and verify failure

Run:

go test ./internal/stats -run TestProjectDailyMileageSelectsCandidateAndMarksSource -count=1

Expected: FAIL because ProjectDailyMileage is undefined.

  • Step 3: Implement projector

Add to go/vehicle-gateway/internal/stats/source_mileage.go:

const maxSelectedDailyMileageKM = 1000

func ProjectDailyMileage(ctx context.Context, exec Execer, vin string, statDate string, protocol envelope.Protocol) error {
	if strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" {
		return nil
	}
	if _, err := exec.ExecContext(ctx, clearSelectedSourceSQL, vin, statDate, string(protocol)); err != nil {
		return err
	}
	_, err := exec.ExecContext(ctx, projectDailyMileageSQL,
		vin,
		statDate,
		string(protocol),
		maxSelectedDailyMileageKM,
		vin,
		statDate,
		string(protocol),
	)
	return err
}

const clearSelectedSourceSQL = `
UPDATE vehicle_daily_mileage_source
SET is_selected = 0
WHERE vin = ? AND stat_date = ? AND protocol = ?
`

const projectDailyMileageSQL = `
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)
SELECT
  s.vin,
  s.stat_date,
  s.protocol,
  s.daily_mileage_km,
  s.first_total_mileage_km,
  s.latest_total_mileage_km,
  s.source_key,
  s.phone,
  s.source_endpoint,
  s.sample_count
FROM vehicle_daily_mileage_source s
LEFT JOIN vehicle_data_source ds
  ON ds.protocol = s.protocol AND ds.source_ip = s.source_ip
WHERE s.vin = ?
  AND s.stat_date = ?
  AND s.protocol = ?
  AND s.quality_status = 'OK'
  AND s.daily_mileage_km BETWEEN 0 AND ?
  AND COALESCE(ds.enabled, 1) = 1
ORDER BY COALESCE(ds.trust_priority, 100),
         s.sample_count DESC,
         s.latest_event_time DESC,
         s.source_key ASC
LIMIT 1
ON DUPLICATE KEY UPDATE
  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),
  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
`

Then add a third exec in ProjectDailyMileage to mark the selected candidate:

const markSelectedSourceSQL = `
UPDATE vehicle_daily_mileage_source s
JOIN vehicle_daily_mileage m
  ON m.vin = s.vin
 AND m.stat_date = s.stat_date
 AND m.protocol = s.protocol
 AND m.trusted_source_key = s.source_key
SET s.is_selected = 1
WHERE s.vin = ? AND s.stat_date = ? AND s.protocol = ?
`

Update the test to expect three calls and assert markSelectedSourceSQL contains SET s.is_selected = 1.

  • Step 4: Run projector tests and verify pass

Run:

go test ./internal/stats -run TestProjectDailyMileageSelectsCandidateAndMarksSource -count=1

Expected: PASS.

  • Step 5: Commit projector

Run:

git add go/vehicle-gateway/internal/stats/source_mileage.go go/vehicle-gateway/internal/stats/source_mileage_test.go
git commit -m "feat(stats): project elected mileage source"

Task 4: Rewire Real-Time Stat Writer To Candidates

Files:

  • Modify: go/vehicle-gateway/internal/stats/daily_metric.go
  • Modify: go/vehicle-gateway/internal/stats/daily_metric_test.go

Interfaces:

  • Consumes: UpsertDataSource

  • Consumes: UpsertSourceMileage

  • Consumes: ProjectDailyMileage

  • Produces: Writer.Append writes source metadata, candidate row, and final projection.

  • Step 1: Write failing Writer.Append flow test

Add to go/vehicle-gateway/internal/stats/daily_metric_test.go:

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)
	}
}
  • Step 2: Run flow test and verify failure

Run:

go test ./internal/stats -run TestWriterAppendWritesSourceCandidateAndProjection -count=1

Expected: FAIL because Writer.Append still calls the old direct final upsert.

  • Step 3: Rewrite Writer.Append

Modify Writer.Append in go/vehicle-gateway/internal/stats/daily_metric.go:

func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
	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
		}
		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)
	}
	return nil
}

Keep upsertDailyMileageSQL only if older tests still reference it; remove it after tests prove no code path uses it.

  • Step 4: Update schema bootstrap

Modify EnsureSchema in go/vehicle-gateway/internal/stats/daily_metric.go:

func (w *Writer) EnsureSchema(ctx context.Context) error {
	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) {
			return err
		}
	}
	return nil
}
  • Step 5: Run stats tests

Run:

go test ./internal/stats -count=1

Expected: PASS.

  • Step 6: Commit realtime writer rewire

Run:

git add go/vehicle-gateway/internal/stats/daily_metric.go go/vehicle-gateway/internal/stats/daily_metric_test.go
git commit -m "feat(stats): write realtime mileage candidates"

Task 5: Backfill Candidate Rows And Final Projection

Files:

  • Modify: go/vehicle-gateway/cmd/stats-backfill/main.go
  • Modify: go/vehicle-gateway/cmd/stats-backfill/main_test.go

Interfaces:

  • Consumes: UpsertDataSource

  • Consumes: UpsertSourceMileage

  • Consumes: ProjectDailyMileage

  • Produces: backfill can rebuild source candidates and final result for a date/protocol range.

  • Step 1: Write failing backfill election test

Modify go/vehicle-gateway/cmd/stats-backfill/main_test.go to keep the existing TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump and add:

func TestDailySourceLastBuildsCandidateKeysBySourceIP(t *testing.T) {
	sourceA := dailySourceLast{
		VIN:            "LA9GG64L7PBAF4001",
		SourceKey:      normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:20215"),
		Phone:          "13307765812",
		SourceEndpoint: "115.231.168.135:20215",
		TotalKM:        4123.9,
	}
	sourceB := dailySourceLast{
		VIN:            "LA9GG64L7PBAF4001",
		SourceKey:      normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:42630"),
		Phone:          "13307765812",
		SourceEndpoint: "115.231.168.135:42630",
		TotalKM:        4100.8,
	}
	if sourceA.SourceKey != sourceB.SourceKey {
		t.Fatalf("same source IP should produce same source key: %q vs %q", sourceA.SourceKey, sourceB.SourceKey)
	}
}
  • Step 2: Run backfill tests and verify failure

Run:

go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP' -count=1

Expected: FAIL until normalizedSourceKey accepts protocol and device fields.

  • Step 3: Update source key normalization in backfill

Change normalizedSourceKey in go/vehicle-gateway/cmd/stats-backfill/main.go:

func normalizedSourceKey(protocol string, phone string, deviceID string, endpoint string) string {
	sourceIP := stats.NormalizeSourceIP(endpoint)
	return stats.SourceKey(envelope.Protocol(protocol), phone, deviceID, sourceIP)
}

Update all calls:

sourceKey := normalizedSourceKey(string(protocol), phone, "", sourceEndpoint)
  • Step 4: Change backfill write path to candidates

Replace final-only writeAggregates with candidate write and projection:

func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*metricAgg, batchSize int) (int64, error) {
	var written int64
	for _, agg := range aggregates {
		identity := stats.SourceIdentity{
			Protocol:       agg.Protocol,
			SourceIP:       stats.NormalizeSourceIP(agg.SourceEndpoint),
			SourceEndpoint: agg.SourceEndpoint,
		}
		if err := stats.UpsertDataSource(ctx, db, identity, agg.LatestEventTime); err != nil {
			return written, err
		}
		candidate := stats.SourceMileageSample{
			VIN:             agg.VIN,
			StatDate:        agg.Date,
			Protocol:        agg.Protocol,
			SourceKey:       agg.SourceKey,
			SourceIP:        identity.SourceIP,
			SourceEndpoint:  agg.SourceEndpoint,
			Phone:           agg.Phone,
			DeviceID:        agg.DeviceID,
			FirstTotalKM:    agg.FirstKM,
			LatestTotalKM:   agg.LatestKM,
			DailyKM:         agg.LatestKM - agg.FirstKM,
			SampleCount:     agg.Count,
			FirstEventTime:  agg.FirstEventTime,
			LatestEventTime: agg.LatestEventTime,
			QualityStatus:   stats.QualityOK,
			QualityReason:   "same_source_previous_day",
		}
		if candidate.DailyKM < 0 || candidate.DailyKM > 1000 {
			candidate.QualityStatus = stats.QualityInvalidDelta
			candidate.QualityReason = "outside_daily_range"
		}
		if err := stats.UpsertSourceMileage(ctx, db, candidate); err != nil {
			return written, err
		}
		if err := stats.ProjectDailyMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
			return written, err
		}
		written++
	}
	return written, nil
}

Extend metricAgg with DeviceID, FirstEventTime, and LatestEventTime.

  • Step 5: Preserve every source candidate

Update buildLastDiffAggregates so it writes one aggregate per source candidate when same-source previous baseline exists, not just one chosen source.

Use this key:

key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey

When a current source has no previous source, create a candidate with:

QualityStatus: stats.QualityNoPreviousBaseline
QualityReason: "missing_previous_source"
FirstKM: currentRow.TotalKM
LatestKM: currentRow.TotalKM
DailyKM: 0

Do not project invalid candidates; projection SQL already filters quality_status = 'OK'.

  • Step 6: Run backfill tests

Run:

go test ./cmd/stats-backfill ./internal/stats -count=1

Expected: PASS.

  • Step 7: Commit backfill candidate support

Run:

git add go/vehicle-gateway/cmd/stats-backfill/main.go go/vehicle-gateway/cmd/stats-backfill/main_test.go
git commit -m "feat(stats): backfill mileage source candidates"

Task 6: API Query Compatibility And Candidate Audit Hook

Files:

  • Modify: go/vehicle-gateway/internal/stats/query.go
  • Modify: go/vehicle-gateway/internal/stats/query_test.go

Interfaces:

  • Produces: existing /api/stats/daily-metrics response remains backward-compatible.

  • Produces: selected source metadata is visible in final rows when present.

  • Step 1: Write failing query test for selected source fields

Add to go/vehicle-gateway/internal/stats/query_test.go:

func TestMetricQueryReturnsSelectedSourceFields(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New() error = %v", err)
	}
	defer db.Close()

	rows := sqlmock.NewRows([]string{
		"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",
	}).AddRow(
		"LA9GG64L7PBAF4001", "2026-07-08", "JT808", 23.1,
		4100.8, 4123.9,
		"JT808:13307765812@115.231.168.135", "13307765812", "115.231.168.135:20215",
		1, "2026-07-08 13:30:57",
	)
	mock.ExpectQuery("SELECT vin, stat_date, protocol").WillReturnRows(rows)

	repo := NewMetricRepository(db)
	got, err := repo.Query(context.Background(), MetricQuery{
		VIN:      "LA9GG64L7PBAF4001",
		Protocol: "JT808",
		Limit:    1,
	})
	if err != nil {
		t.Fatalf("Query() error = %v", err)
	}
	if got[0].TrustedSourceKey == nil || *got[0].TrustedSourceKey != "JT808:13307765812@115.231.168.135" {
		t.Fatalf("trusted source = %#v", got[0].TrustedSourceKey)
	}
}
  • Step 2: Run query test and verify failure

Run:

go test ./internal/stats -run TestMetricQueryReturnsSelectedSourceFields -count=1

Expected: FAIL because MetricRow does not expose trusted source fields.

  • Step 3: Extend MetricRow and SQL

Modify MetricRow in go/vehicle-gateway/internal/stats/query.go:

TrustedSourceKey      *string `json:"trusted_source_key,omitempty"`
TrustedPhone          *string `json:"trusted_phone,omitempty"`
TrustedSourceEndpoint *string `json:"trusted_source_endpoint,omitempty"`

Modify selected columns:

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

Scan into sql.NullString values and set pointers when valid.

  • Step 4: Run query tests

Run:

go test ./internal/stats -run 'TestMetricQuery|TestMetricHandler' -count=1

Expected: PASS.

  • Step 5: Commit query compatibility

Run:

git add go/vehicle-gateway/internal/stats/query.go go/vehicle-gateway/internal/stats/query_test.go
git commit -m "feat(stats): expose selected mileage source"

Task 7: Deploy, Rebuild, And Verify Production Data

Files:

  • Uses built binaries from go/vehicle-gateway/cmd/stat-writer
  • Uses built binaries from go/vehicle-gateway/cmd/stats-backfill

Interfaces:

  • Consumes: ECS env files at /opt/lingniu-go-native/env/stat-writer.env and /opt/lingniu-go-native/env/history-writer.env

  • Produces: source tables and final table populated on ECS.

  • Step 1: Run full test suite locally

Run:

cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
go test ./... -count=1

Expected: PASS for all packages.

  • Step 2: Build Linux binaries

Run:

mkdir -p ../../outputs/multi_source_stats_20260708/bin
GOOS=linux GOARCH=amd64 go build -trimpath -ldflags='-s -w' -o ../../outputs/multi_source_stats_20260708/bin/stat-writer ./cmd/stat-writer
GOOS=linux GOARCH=amd64 go build -trimpath -ldflags='-s -w' -o ../../outputs/multi_source_stats_20260708/bin/stats-backfill ./cmd/stats-backfill

Expected: both binaries exist in ../../outputs/multi_source_stats_20260708/bin.

  • Step 3: Upload and restart stat writer

Run:

scp -q ../../outputs/multi_source_stats_20260708/bin/stat-writer root@115.29.187.205:/tmp/stat-writer.new
scp -q ../../outputs/multi_source_stats_20260708/bin/stats-backfill root@115.29.187.205:/tmp/stats-backfill.new
ssh root@115.29.187.205 'set -e
stamp=$(date +%Y%m%d%H%M%S)
cp /opt/lingniu-go-native/current/stat-writer /opt/lingniu-go-native/current/stat-writer.bak.$stamp
cp /opt/lingniu-go-native/current/stats-backfill /opt/lingniu-go-native/current/stats-backfill.bak.$stamp || true
install -m 755 /tmp/stat-writer.new /opt/lingniu-go-native/current/stat-writer
install -m 755 /tmp/stats-backfill.new /opt/lingniu-go-native/current/stats-backfill
rm -f /tmp/stat-writer.new /tmp/stats-backfill.new
systemctl restart lingniu-go-stat-writer.service
systemctl is-active lingniu-go-stat-writer.service
journalctl -u lingniu-go-stat-writer.service -n 20 --no-pager'

Expected: service status is active.

  • Step 4: Rebuild 2026-07-08 JT808

Run:

ssh root@115.29.187.205 'BACKFILL_ENV_FILES=/opt/lingniu-go-native/env/stat-writer.env,/opt/lingniu-go-native/env/history-writer.env BACKFILL_METHOD=last_diff BACKFILL_DATE_FROM=2026-07-08 BACKFILL_DATE_TO=2026-07-08 BACKFILL_PROTOCOLS=JT808 BACKFILL_DRY_RUN=false BACKFILL_RESET=true /opt/lingniu-go-native/current/stats-backfill'

Expected output includes stats backfill complete and non-zero candidate/final counts.

  • Step 5: Verify anomaly VIN

Run:

curl -fsS 'http://115.29.187.205:20200/api/stats/daily-metrics?vin=LA9GG64L7PBAF4001&protocol=JT808&dateFrom=2026-07-08&dateTo=2026-07-08&limit=10&includeTotal=true'

Expected JSON contains:

{
  "vin": "LA9GG64L7PBAF4001",
  "daily_mileage_km": 23.1,
  "first_total_mileage_km": 4100.8,
  "latest_total_mileage_km": 4123.9
}
  • Step 6: Verify candidates exist

Run a temporary MySQL query binary or existing inspection helper to execute:

SELECT vin, stat_date, protocol, source_key, source_ip, daily_mileage_km, quality_status, is_selected
FROM vehicle_daily_mileage_source
WHERE vin = 'LA9GG64L7PBAF4001'
  AND stat_date = '2026-07-08'
  AND protocol = 'JT808'
ORDER BY source_key;

Expected:

  • At least one selected source row with is_selected = 1.

  • At least one non-selected row when alternate 424xx source data exists.

  • Non-selected invalid rows remain available for audit.

  • Step 7: Verify stat writer health

Run:

ssh root@115.29.187.205 "curl -fsS http://127.0.0.1:20213/metrics | grep -E 'vehicle_stat_(kafka_lag|writes)' | tail -40"

Expected:

  • vehicle_stat_kafka_lag is 0 for GB32960, JT808, and Yutong MQTT fields topics.

  • vehicle_stat_writes_total{status="ok"} is increasing.

  • Step 8: Commit deployment-ready changes

Run:

git status --short
git add go/vehicle-gateway/internal/stats go/vehicle-gateway/cmd/stats-backfill
git commit -m "feat(stats): persist multi-source mileage candidates"
git push origin go

Expected: branch go is pushed successfully.

Self-Review

  • Spec coverage: source metadata, source candidates, final projection, source election, API compatibility, backfill, and production verification are each covered by at least one task.
  • Placeholder scan: this plan contains no placeholder markers or open-ended implementation instructions.
  • Type consistency: source identity flows from SourceIdentity to SourceMileageSample; final projection always consumes candidate rows and updates vehicle_daily_mileage.
Reference in New Issue View Git Blame Copy Permalink