chore: commit mileage updates and accumulated project artifacts

2026-09-21 09:15:54 +08:00
parent 6cc872ceb6
commit 8b32ae2988
767 changed files with 572885 additions and 17 deletions
@@ -0,0 +1,105 @@
+// Apply a reviewed, backed-up stale-baseline plan and rebuild its projections atomically.
+package main
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"flag"
+	"fmt"
+	_ "github.com/go-sql-driver/mysql"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats"
+	"math"
+	"os"
+	"sort"
+	"strconv"
+	"time"
+)
+
+type change struct {
+	Before struct {
+		VIN       string `json:"vin"`
+		Protocol  string `json:"protocol"`
+		Date      string `json:"stat_date"`
+		Source    string `json:"source_key"`
+		First     string `json:"first_total_mileage_km"`
+		FirstTime string `json:"first_event_time"`
+		Total     string `json:"latest_total_mileage_km"`
+		LastTime  string `json:"latest_event_time"`
+		Daily     string `json:"daily_mileage_km"`
+	} `json:"before"`
+	Baseline struct {
+		Total string `json:"total"`
+		Time  string `json:"time"`
+	} `json:"baseline"`
+	Daily  string `json:"daily"`
+	Status string `json:"status"`
+	Reason string `json:"reason"`
+}
+
+func main() {
+	input := flag.String("input", "", "reviewed JSON plan")
+	apply := flag.Bool("apply", false, "apply plan and projections in one transaction")
+	manifest := flag.String("backup-manifest", "", "verified backup manifest required for apply")
+	flag.Parse()
+	data, err := os.ReadFile(*input)
+	must(err)
+	var changes []change
+	must(json.Unmarshal(data, &changes))
+	if !*apply {
+		fmt.Printf("planned=%d dry_run=true\n", len(changes))
+		return
+	}
+	backup, err := os.ReadFile(*manifest)
+	must(err)
+	if !json.Valid(backup) {
+		panic("invalid backup manifest")
+	}
+	db, err := sql.Open("mysql", os.Getenv("MYSQL_DSN"))
+	must(err)
+	defer db.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+	defer cancel()
+	tx, err := db.BeginTx(ctx, nil)
+	must(err)
+	defer tx.Rollback()
+	affected := map[string]change{}
+	for _, c := range changes {
+		b := c.Before
+		total, stamp, found, err := stats.LookupLatestSourceBaselineBefore(ctx, tx, b.VIN, b.Date, envelope.Protocol(b.Protocol), b.Source)
+		must(err)
+		expected, err := strconv.ParseFloat(c.Baseline.Total, 64)
+		must(err)
+		if !found || math.Abs(total-expected) > 0.0001 || stamp.Format("2006-01-02 15:04:05") != c.Baseline.Time {
+			panic(fmt.Sprintf("baseline changed: %s %s", b.VIN, b.Date))
+		}
+		result, err := tx.ExecContext(ctx, `UPDATE vehicle_daily_mileage_source SET first_total_mileage_km=?,first_event_time=?,daily_mileage_km=?,quality_status=?,quality_reason=?
+  WHERE vin=? AND stat_date=? AND protocol=? AND source_key=?
+  AND first_total_mileage_km <=> ? AND first_event_time <=> ? AND latest_total_mileage_km=?
+  AND latest_event_time=? AND daily_mileage_km=? AND quality_status='OK'`, c.Baseline.Total, c.Baseline.Time, c.Daily, c.Status, c.Reason, b.VIN, b.Date, b.Protocol, b.Source, b.First, b.FirstTime, b.Total, b.LastTime, b.Daily)
+		must(err)
+		n, err := result.RowsAffected()
+		must(err)
+		if n != 1 {
+			panic(fmt.Sprintf("concurrent change: %s %s %s", b.VIN, b.Date, b.Source))
+		}
+		affected[b.Date+"|"+b.VIN+"|"+b.Protocol] = c
+	}
+	keys := make([]string, 0, len(affected))
+	for key := range affected {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	for _, key := range keys {
+		b := affected[key].Before
+		must(stats.ProjectDailyMileage(ctx, tx, b.VIN, b.Date, envelope.Protocol(b.Protocol)))
+	}
+	must(tx.Commit())
+	must(json.NewEncoder(os.Stdout).Encode(map[string]any{"updatedSources": len(changes), "reprojectedDays": len(keys), "committed": true}))
+}
+func must(err error) {
+	if err != nil {
+		panic(err)
+	}
+}
@@ -914,10 +914,41 @@ func latestHistoricalSourceRow(history map[sourceHistoryID]dailySourceLast, curr
 }

 func resolvePreviousSourceRow(ctx context.Context, db *sql.DB, date string, protocol envelope.Protocol, history map[sourceHistoryID]dailySourceLast, current dailySourceLast) (dailySourceLast, bool, error) {
-	if previous, ok := latestHistoricalSourceRow(history, current); ok {
-		return previous, true, nil
+	previous, found := latestHistoricalSourceRow(history, current)
+	if protocol != envelope.ProtocolGB32960 {
+		// Retained raw history can be older than an already persisted daily
+		// boundary. Always compare the same-source durable baseline as well.
+		durable, durableFound, err := queryDurablePreviousSourceRow(ctx, db, date, protocol, current)
+		if err != nil {
+			return dailySourceLast{}, false, err
+		}
+		if durableFound && (!found || durable.TS.After(previous.TS)) {
+			return durable, true, nil
+		}
+		return previous, found, nil
 	}
-	return queryDurablePreviousSourceRow(ctx, db, date, protocol, current)
+	// A returning GB32960 feed must not recount another feed's vehicle mileage.
+	day, err := time.ParseInLocation("2006-01-02", date, current.TS.Location())
+	if err != nil {
+		return dailySourceLast{}, false, err
+	}
+	for _, row := range history {
+		if row.VIN != current.VIN || !row.TS.Before(day) || row.TotalKM <= 0 {
+			continue
+		}
+		if !found || row.TS.After(previous.TS) || (row.TS.Equal(previous.TS) && row.SourceKey < previous.SourceKey) {
+			previous, found = row, true
+		}
+	}
+	// Durable projections may be newer than retained raw history.
+	durable, durableFound, err := queryDurablePreviousSourceRow(ctx, db, date, protocol, current)
+	if err != nil {
+		return dailySourceLast{}, false, err
+	}
+	if durableFound && (!found || durable.TS.In(current.TS.Location()).Format("2006-01-02") >= previous.TS.In(current.TS.Location()).Format("2006-01-02")) {
+		return durable, true, nil
+	}
+	return previous, found, nil
 }

 func aggregateFromDailySource(date string, protocol envelope.Protocol, current dailySourceLast, previous dailySourceLast, hasPrevious bool) *metricAgg {
@@ -2,6 +2,7 @@ package main

 import (
 	"context"
+	"math"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -568,6 +569,9 @@ func TestAddRealtimeLocationFallbackReusesAggregateHistoryAcrossEmptyDays(t *tes
 		WithArgs("YUTONG_MQTT", "2026-07-11", "2026-07-11").
 		WillReturnRows(sqlmock.NewRows([]string{"vin", "peer", "total_mileage_event_time", "total_mileage_km"}).
 			AddRow(vin, endpoint, dayThreeTS, 120.0))
+	mysqlMock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs(vin, "2026-07-11", "YUTONG_MQTT", sourceKey).
+		WillReturnRows(sqlmock.NewRows([]string{"total", "time"}))

 	aggregates := map[string]*metricAgg{
 		vin + "|2026-07-09|YUTONG_MQTT|" + sourceKey: {
@@ -1322,3 +1326,59 @@ func TestBackfillEnvFilesUsesExistingDefaults(t *testing.T) {
 		t.Fatalf("env files = %q, want %q", got, want)
 	}
 }
+
+func TestReturningGB32960FeedUsesNewerVehicleHistory(t *testing.T) {
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	current := dailySourceLast{VIN: "LB9A32A26R0LS1055", SourceKey: "old-feed", FirstTotalKM: 75429.6, TotalKM: 75429.6, FirstTS: time.Date(2026, 9, 15, 11, 0, 0, 0, loc), TS: time.Date(2026, 9, 15, 11, 27, 37, 0, loc)}
+	old := current
+	old.TotalKM = 40386.7
+	old.TS = time.Date(2026, 4, 24, 12, 14, 51, 0, loc)
+	recent := current
+	recent.SourceKey = "active-feed"
+	recent.TS = time.Date(2026, 9, 9, 19, 9, 35, 0, loc)
+	history := map[sourceHistoryID]dailySourceLast{{VIN: current.VIN, SourceKey: old.SourceKey}: old, {VIN: current.VIN, SourceKey: recent.SourceKey}: recent}
+	previous, found, err := resolvePreviousSourceRow(context.Background(), nil, "2026-09-15", envelope.ProtocolGB32960, history, current)
+	if err != nil || !found || !previous.TS.Equal(recent.TS) {
+		t.Fatalf("previous=%+v found=%v err=%v", previous, found, err)
+	}
+	agg := aggregateFromDailySource("2026-09-15", envelope.ProtocolGB32960, current, previous, found)
+	if agg.LatestKM-agg.FirstKM != 0 || agg.QualityStatus != stats.QualityOK {
+		t.Fatalf("%+v", agg)
+	}
+	// Independent GPS terminal counters must keep their own source baseline.
+	previous, found, err = resolvePreviousSourceRow(context.Background(), nil, "2026-09-15", envelope.ProtocolJT808, history, current)
+	if err != nil || !found || previous.TotalKM != 40386.7 {
+		t.Fatalf("JT808 mixed source counters: %+v %v", previous, err)
+	}
+}
+
+func TestBackfillPrefersNewerDurableSameSourceBaseline(t *testing.T) {
+	for _, protocol := range []envelope.Protocol{envelope.ProtocolJT808, envelope.ProtocolYutongMQTT} {
+		t.Run(string(protocol), func(t *testing.T) {
+			db, mock, err := sqlmock.New()
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer db.Close()
+			loc := time.FixedZone("Asia/Shanghai", 8*3600)
+			current := dailySourceLast{VIN: "LMRKH9AC8R1004109", SourceKey: "same-terminal", TotalKM: 11653.2, FirstTotalKM: 11653.2, TS: time.Date(2026, 9, 16, 12, 56, 12, 0, loc), FirstTS: time.Date(2026, 9, 16, 12, 50, 0, 0, loc)}
+			old := current
+			old.TotalKM = 10.4
+			old.TS = time.Date(2026, 7, 13, 9, 54, 36, 0, loc)
+			durableTime := time.Date(2026, 9, 4, 15, 13, 15, 0, loc)
+			history := map[sourceHistoryID]dailySourceLast{{VIN: current.VIN, SourceKey: current.SourceKey}: old}
+			mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).WithArgs(current.VIN, "2026-09-16", string(protocol), current.SourceKey).WillReturnRows(sqlmock.NewRows([]string{"total", "time"}).AddRow(11653.1, durableTime))
+			previous, found, err := resolvePreviousSourceRow(context.Background(), db, "2026-09-16", protocol, history, current)
+			if err != nil || !found || !previous.TS.Equal(durableTime) {
+				t.Fatalf("baseline=%+v found=%v err=%v", previous, found, err)
+			}
+			agg := aggregateFromDailySource("2026-09-16", protocol, current, previous, found)
+			if math.Abs(agg.LatestKM-agg.FirstKM-0.1) > 0.0001 {
+				t.Fatalf("stale baseline counted again: %+v", agg)
+			}
+			if err := mock.ExpectationsWereMet(); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}