feat(stats): add vehicle data source metadata

.superpowers/sdd/task-1-report.md

@@ -0,0 +1,54 @@
+# Task 1 Report: Source Metadata Schema And Identity Helpers
+
+## Outcome
+- Implemented Task 1 in `go/vehicle-gateway/internal/stats`.
+- Added source metadata schema, identity helpers, and focused tests.
+
+## RED Evidence
+Initial focused test run failed as expected because the new helpers did not exist yet:
+
+```bash
+cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
+go test ./internal/stats -run 'TestNormalizeSourceIP|TestNewSourceIdentity|TestUpsertDataSource' -count=1
+```
+
+Result:
+- `undefined: NormalizeSourceIP`
+- `undefined: NewSourceIdentity`
+- `undefined: SourceIdentity`
+- `undefined: UpsertDataSource`
+
+## GREEN Evidence
+After implementation, the focused tests passed:
+
+```bash
+cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
+go test ./internal/stats -run 'TestNormalizeSourceIP|TestNewSourceIdentity|TestUpsertDataSource' -count=1
+```
+
+Result:
+- `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.927s`
+
+Full package verification also passed:
+
+```bash
+cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
+go test ./internal/stats -count=1
+```
+
+Result:
+- `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.355s`
+
+## Files Changed
+- `go/vehicle-gateway/internal/stats/schema.go`
+- `go/vehicle-gateway/internal/stats/source.go`
+- `go/vehicle-gateway/internal/stats/source_test.go`
+
+## Notes
+- `DataSourceTableSQL` was added verbatim to `schema.go` for the future schema wiring step.
+- `UpsertDataSource` follows the brief exactly, including the nil exec panic, empty-source short circuit, and SQL shape.
+
+## Self-Review
+- The implementation is tightly scoped to Task 1.
+- The new tests cover IP normalization, identity construction, and the SQL shape for the upsert helper.
+- No additional concerns at this stage.

go/vehicle-gateway/internal/stats/schema.go

@@ -22,3 +22,21 @@ var DailyMileageAlterSQL = []string{
 	"ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_phone VARCHAR(32) NULL",
 	"ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_source_endpoint VARCHAR(128) NULL",
 }
+
+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)
+)`

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

@@ -0,0 +1,68 @@
+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
+`

go/vehicle-gateway/internal/stats/source_test.go

@@ -0,0 +1,79 @@
+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)
+		}
+	}
+}