lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill/main.go

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"

	_ "github.com/go-sql-driver/mysql"
	_ "github.com/taosdata/driver-go/v3/taosWS"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/history"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats"
)

type config struct {
	MySQLDSN          string
	TDengineDriver    string
	TDengineDSN       string
	TDengineDatabase  string
	DateFrom          string
	DateTo            string
	Protocols         []envelope.Protocol
	Method            string
	Limit             int
	DryRun            bool
	Reset             bool
	Debug             bool
	EventTimeFullScan bool
	StationaryCarry   bool
	JT808GPSFallback  bool
	ProgressEvery     int64
	BaselineLookback  int
	Location          *time.Location
}

type rawFrameRow struct {
	Protocol       envelope.Protocol
	VIN            string
	Phone          string
	DeviceID       string
	SourceEndpoint string
	EventID        string
	MessageID      string
	EventTimeMS    int64
	ReceivedAtMS   int64
	ParsedJSON     string
	RawText        string
}

type metricAgg struct {
	VIN             string
	Date            string
	Protocol        envelope.Protocol
	FirstKM         float64
	LatestKM        float64
	Count           int64
	SourceKey       string
	Phone           string
	DeviceID        string
	SourceEndpoint  string
	SourceCode      string
	PlatformName    string
	SourceKind      string
	FirstEventTime  time.Time
	LatestEventTime time.Time
	QualityStatus   string
	QualityReason   string
}

type dailySourceLast struct {
	VIN            string
	SourceKey      string
	Phone          string
	DeviceID       string
	SourceEndpoint string
	SourceCode     string
	PlatformName   string
	SourceKind     string
	FirstTS        time.Time
	TS             time.Time
	FirstTotalKM   float64
	TotalKM        float64
	RawSampleCount int64
}

type sourceHistoryID struct {
	VIN       string
	SourceKey string
}

var defaultBackfillEnvFiles = []string{
	"/opt/lingniu-go-native/env/base.env",
	"/opt/lingniu-go-native/env/stat-writer.env",
}

func main() {
	envFiles := backfillEnvFiles(os.Getenv("BACKFILL_ENV_FILES"), os.Getenv("BACKFILL_ENV_FILE"), defaultBackfillEnvFiles)
	if err := loadEnvFiles(envFiles); err != nil {
		fail("load env file", err)
	}
	cfg, err := loadConfig()
	if err != nil {
		fail("load config", err)
	}
	ctx := context.Background()

	td, err := sql.Open(cfg.TDengineDriver, cfg.TDengineDSN)
	if err != nil {
		fail("open tdengine", err)
	}
	defer td.Close()
	if err := td.PingContext(ctx); err != nil {
		fail("ping tdengine", err)
	}

	mysqlDB, err := sql.Open("mysql", cfg.MySQLDSN)
	if err != nil {
		fail("open mysql", err)
	}
	defer mysqlDB.Close()
	if err := mysqlDB.PingContext(ctx); err != nil {
		fail("ping mysql", err)
	}

	schemaWriter := stats.NewWriter(mysqlDB, cfg.Location)
	if err := schemaWriter.EnsureSchema(ctx); err != nil {
		fail("ensure schema", err)
	}
	if cfg.Reset && !cfg.DryRun {
		deleted, err := resetStats(ctx, mysqlDB, cfg)
		if err != nil {
			fail("reset stats", err)
		}
		slog.Info("reset stats rows", "deleted", deleted)
	}
	if cfg.Method == "last_diff" {
		aggregates, err := buildLastDiffAggregates(ctx, mysqlDB, td, cfg)
		if err != nil {
			fail("build last-diff aggregates", err)
		}
		fallbacks, err := addRealtimeLocationFallbackAggregates(ctx, mysqlDB, td, cfg, aggregates)
		if err != nil {
			fail("build realtime-location fallback aggregates", err)
		}
		gpsFallbacks, err := addJT808GPSCoordinateFallbackAggregates(ctx, mysqlDB, td, cfg, aggregates)
		if err != nil {
			fail("build JT808 GPS-coordinate fallback aggregates", err)
		}
		var written int64
		var normalized int
		if !cfg.DryRun {
			written, err = writeAggregates(ctx, mysqlDB, aggregates, 500)
			if err != nil {
				fail("write aggregates", err)
			}
			normalized, err = normalizeHistoricalPlatformSources(ctx, mysqlDB, cfg)
			if err != nil {
				fail("normalize historical platform sources", err)
			}
		}
		slog.Info("stats backfill complete", "method", cfg.Method, "dry_run", cfg.DryRun, "aggregates", len(aggregates), "realtimeLocationFallbacks", fallbacks, "jt808GPSFallbacks", gpsFallbacks, "written", written, "platformSourcesNormalized", normalized)
		return
	}

	rows, err := queryRawFrames(ctx, td, cfg)
	if err != nil {
		fail("query raw frames", err)
	}
	defer rows.Close()

	aggregates := map[string]*metricAgg{}
	var scanned, chunked, parsed, sampled, written, skipped int64
	for rows.Next() {
		scanned++
		row, err := scanRawFrame(rows)
		if err != nil {
			fail("scan raw frame", err)
		}
		text := strings.TrimSpace(row.ParsedJSON)
		if isChunkManifest(text) {
			chunked++
			text, err = readChunkedPayload(ctx, td, cfg.TDengineDatabase, row.EventID, "parsed_fields")
			if err != nil {
				slog.Warn("skip chunked parsed fields", "event_id", row.EventID, "error", err)
				skipped++
				continue
			}
		}
		fields := fieldsForStats(row.Protocol, row.VIN, text)
		if len(fields) == 0 && row.Protocol == envelope.ProtocolYutongMQTT {
			fields = fieldsForStats(row.Protocol, row.VIN, row.RawText)
		}
		if cfg.Debug && scanned <= 5 {
			slog.Info("debug raw frame",
				"protocol", row.Protocol,
				"vin", row.VIN,
				"message_id", row.MessageID,
				"parsed_len", len(text),
				"mileage_keys", mileageKeys(row.Protocol),
				"extracted_fields", fields)
		}
		if len(fields) == 0 {
			skipped++
			continue
		}
		parsed++
		env := envelope.FrameEnvelope{
			EventID:        row.EventID,
			Protocol:       row.Protocol,
			MessageID:      row.MessageID,
			VIN:            row.VIN,
			Phone:          row.Phone,
			DeviceID:       row.DeviceID,
			SourceEndpoint: row.SourceEndpoint,
			EventTimeMS:    row.EventTimeMS,
			ReceivedAtMS:   row.ReceivedAtMS,
			Fields:         fields,
			ParsedFields:   fields,
			ParseStatus:    envelope.ParseOK,
		}
		samples, err := stats.SamplesFromEnvelope(env, cfg.Location)
		if err != nil {
			slog.Warn("skip invalid sample", "event_id", row.EventID, "error", err)
			skipped++
			continue
		}
		if len(samples) == 0 {
			skipped++
			continue
		}
		sampled += int64(len(samples))
		addSamples(aggregates, samples)
		if cfg.ProgressEvery > 0 && scanned%cfg.ProgressEvery == 0 {
			slog.Info("stats backfill progress", "scanned", scanned, "sampled", sampled, "aggregates", len(aggregates), "skipped", skipped)
		}
	}
	if err := rows.Err(); err != nil {
		fail("iterate raw frames", err)
	}
	if !cfg.DryRun {
		var err error
		written, err = writeAggregates(ctx, mysqlDB, aggregates, 500)
		if err != nil {
			fail("write aggregates", err)
		}
	}
	slog.Info("stats backfill complete", "dry_run", cfg.DryRun, "scanned", scanned, "parsed", parsed, "chunked", chunked, "sampled", sampled, "aggregates", len(aggregates), "written", written, "skipped", skipped)
}

func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample) {
	for _, sample := range samples {
		key := sample.VIN + "|" + sample.StatDate + "|" + string(sample.Protocol) + "|" + sample.SourceKey
		agg, ok := aggregates[key]
		if !ok {
			aggregates[key] = &metricAgg{
				VIN:             sample.VIN,
				Date:            sample.StatDate,
				Protocol:        sample.Protocol,
				FirstKM:         sample.TotalMileageKM,
				LatestKM:        sample.TotalMileageKM,
				Count:           1,
				SourceKey:       sample.SourceKey,
				Phone:           sample.Phone,
				DeviceID:        sample.DeviceID,
				SourceEndpoint:  sample.SourceEndpoint,
				PlatformName:    sample.PlatformName,
				FirstEventTime:  sample.EventTime,
				LatestEventTime: sample.EventTime,
				QualityStatus:   stats.QualityOK,
				QualityReason:   "current_day_first_sample",
			}
			continue
		}
		if agg.FirstEventTime.IsZero() || sample.EventTime.Before(agg.FirstEventTime) {
			agg.FirstKM = sample.TotalMileageKM
			agg.FirstEventTime = sample.EventTime
		}
		if sample.EventTime.After(agg.LatestEventTime) {
			agg.LatestKM = sample.TotalMileageKM
			agg.LatestEventTime = sample.EventTime
			agg.SourceEndpoint = sample.SourceEndpoint
			if strings.TrimSpace(sample.Phone) != "" {
				agg.Phone = sample.Phone
			}
			if strings.TrimSpace(sample.DeviceID) != "" {
				agg.DeviceID = sample.DeviceID
			}
			if strings.TrimSpace(sample.PlatformName) != "" {
				agg.PlatformName = sample.PlatformName
			}
		}
		agg.Count++
	}
}

func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*metricAgg, batchSize int) (int64, error) {
	if len(aggregates) == 0 {
		return 0, nil
	}
	var written int64
	clearedTargets := map[string]struct{}{}
	for _, agg := range aggregates {
		identity := stats.SourceIdentity{
			Protocol:       agg.Protocol,
			SourceIP:       stats.NormalizeSourceIP(agg.SourceEndpoint),
			SourceEndpoint: agg.SourceEndpoint,
			SourceCode:     agg.SourceCode,
			PlatformName:   agg.PlatformName,
			SourceKind:     agg.SourceKind,
		}
		if strings.TrimSpace(identity.SourceIP) == "" {
			continue
		}
		target := agg.VIN + "|" + agg.Date + "|" + string(agg.Protocol)
		if _, ok := clearedTargets[target]; !ok {
			if err := clearBackfillTargetMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
				return written, err
			}
			clearedTargets[target] = struct{}{}
		}
		if agg.QualityReason != stats.QualityReasonGPSCoordinate {
			if err := stats.UpsertDataSource(ctx, db, identity, agg.LatestEventTime); err != nil {
				return written, err
			}
		}
		dailyKM := stats.DailyMileageFromDayBoundary(agg.FirstKM, agg.LatestKM)
		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,
			PlatformName:    agg.PlatformName,
			FirstTotalKM:    agg.FirstKM,
			LatestTotalKM:   agg.LatestKM,
			DailyKM:         dailyKM,
			SampleCount:     agg.Count,
			FirstEventTime:  agg.FirstEventTime,
			LatestEventTime: agg.LatestEventTime,
			QualityStatus:   agg.QualityStatus,
			QualityReason:   agg.QualityReason,
		}
		if candidate.QualityStatus == "" {
			candidate.QualityStatus = stats.QualityOK
		}
		if candidate.QualityReason == "" {
			candidate.QualityReason = "same_source_previous_day"
		}
		stats.ApplyMileageQualityRules(&candidate)
		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
}

func normalizeHistoricalPlatformSources(ctx context.Context, db *sql.DB, cfg config) (int, error) {
	dates, err := dateRange(cfg.DateFrom, cfg.DateTo)
	if err != nil {
		return 0, err
	}
	normalized := 0
	for _, protocol := range cfg.Protocols {
		if protocol != envelope.ProtocolJT808 {
			continue
		}
		for _, date := range dates {
			count, err := stats.NormalizePlatformSourceMileageForDate(ctx, db, date, protocol)
			if err != nil {
				return normalized, err
			}
			normalized += count
		}
	}
	return normalized, nil
}

func clearBackfillTargetMileage(ctx context.Context, db *sql.DB, vin string, statDate string, protocol envelope.Protocol) error {
	if db == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" || strings.TrimSpace(string(protocol)) == "" {
		return nil
	}
	_, err := db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage_source WHERE vin = ? AND stat_date = ? AND protocol = ?", vin, statDate, string(protocol))
	if err != nil {
		return err
	}
	_, err = db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage WHERE vin = ? AND stat_date = ? AND protocol = ?", vin, statDate, string(protocol))
	return err
}

func buildLastDiffAggregates(ctx context.Context, mysqlDB *sql.DB, tdDB *sql.DB, cfg config) (map[string]*metricAgg, error) {
	targetDates, err := dateRange(cfg.DateFrom, cfg.DateTo)
	if err != nil {
		return nil, err
	}
	aggregates := map[string]*metricAgg{}
	for _, protocol := range cfg.Protocols {
		latestHistory := map[sourceHistoryID]dailySourceLast{}
		preWindow, err := queryPreviousLastSourceRows(ctx, tdDB, cfg, protocol, targetDates[0])
		if err != nil {
			return nil, err
		}
		rememberLatestSourceRows(latestHistory, preWindow)
		slog.Info("pre-window baseline loaded", "protocol", protocol, "before_date", targetDates[0], "vehicles", len(preWindow))
		for _, date := range targetDates {
			current, err := queryDailyLastSourceRows(ctx, tdDB, cfg, protocol, date)
			if err != nil {
				return nil, err
			}
			slog.Info("daily last loaded", "protocol", protocol, "date", date, "vehicles", len(current), "historicalSources", len(latestHistory))
			for vin, currentSources := range current {
				for _, currentRow := range currentSources {
					key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey
					previousRow, hasPrevious, err := resolvePreviousSourceRow(ctx, mysqlDB, date, protocol, latestHistory, currentRow)
					if err != nil {
						return nil, err
					}
					aggregates[key] = aggregateFromDailySource(date, protocol, currentRow, previousRow, hasPrevious)
				}
			}
			// Current-day last values become eligible only for later target dates.
			rememberLatestSourceRows(latestHistory, current)
		}
	}
	return aggregates, nil
}

func addRealtimeLocationFallbackAggregates(ctx context.Context, mysqlDB *sql.DB, _ *sql.DB, cfg config, aggregates map[string]*metricAgg) (int, error) {
	if aggregates == nil {
		return 0, fmt.Errorf("aggregates map is nil")
	}
	targetDates, err := dateRange(cfg.DateFrom, cfg.DateTo)
	if err != nil {
		return 0, err
	}
	var added int
	for _, protocol := range cfg.Protocols {
		if protocol != envelope.ProtocolYutongMQTT {
			continue
		}
		latestHistory := map[sourceHistoryID]dailySourceLast{}
		aggregateRowsByDate := indexAggregateSourceRowsByDate(aggregates, protocol)
		for _, date := range targetDates {
			current, err := queryRealtimeLocationLastRows(ctx, mysqlDB, cfg, protocol, date)
			if err != nil {
				return added, err
			}
			for vin, currentSources := range current {
				for i := range currentSources {
					currentRow := currentSources[i]
					key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey
					if _, exists := aggregates[key]; exists {
						continue
					}
					previousRow, hasPrevious, err := resolvePreviousSourceRow(ctx, mysqlDB, date, protocol, latestHistory, currentRow)
					if err != nil {
						return added, err
					}
					if cfg.StationaryCarry && !hasPrevious {
						continue
					}
					if cfg.StationaryCarry {
						// A sparse Yutong location frame can retain an older
						// realtime odometer than the durable daily baseline.
						// Stationary carry-forward represents an explicit zero
						// day, so both ends must use the latest trusted baseline.
						currentRow.FirstTotalKM = previousRow.TotalKM
						currentRow.TotalKM = previousRow.TotalKM
						currentSources[i] = currentRow
					}
					agg := aggregateFromDailySource(date, protocol, currentRow, previousRow, hasPrevious)
					if cfg.StationaryCarry {
						agg.QualityReason = stats.QualityReasonStationaryCarry
					} else if hasPrevious {
						agg.QualityReason = "realtime_location_fallback_historical_baseline"
					} else {
						agg.QualityReason = "realtime_location_fallback_current_day_first_baseline"
					}
					aggregates[key] = agg
					added++
				}
			}
			rememberLatestSourceRows(latestHistory, aggregateRowsByDate[date])
			rememberLatestSourceRows(latestHistory, current)
			if len(current) > 0 {
				slog.Info("realtime location fallback loaded", "protocol", protocol, "date", date, "vehicles", len(current), "added", added)
			}
		}
	}
	return added, nil
}

type gpsCoordinateBackfillState struct {
	previous        stats.GPSMileagePoint
	firstEventTime  time.Time
	latestEventTime time.Time
	distanceKM      float64
	pointCount      int64
	usableSegments  int64
	badJumps        int64
	longGaps        int64
}

func addJT808GPSCoordinateFallbackAggregates(ctx context.Context, mysqlDB *sql.DB, tdDB *sql.DB, cfg config, aggregates map[string]*metricAgg) (int, error) {
	if !cfg.JT808GPSFallback || tdDB == nil || aggregates == nil || !containsProtocol(cfg.Protocols, envelope.ProtocolJT808) {
		return 0, nil
	}
	existing, err := existingJT808OdometerTargets(ctx, mysqlDB, cfg)
	if err != nil {
		return 0, err
	}
	for _, agg := range aggregates {
		if agg != nil && agg.Protocol == envelope.ProtocolJT808 && agg.QualityReason != stats.QualityReasonGPSCoordinate {
			existing[agg.VIN+"|"+agg.Date] = struct{}{}
		}
	}
	where := []string{
		fmt.Sprintf("ts >= '%s 00:00:00'", quote(cfg.DateFrom)),
		fmt.Sprintf("ts < '%s 00:00:00'", quote(nextDate(cfg.DateTo))),
		"protocol = 'JT808'",
		"vin IS NOT NULL",
		"vin <> ''",
		"longitude BETWEEN -180 AND 180",
		"latitude BETWEEN -90 AND 90",
		"NOT (longitude = 0 AND latitude = 0)",
	}
	sqlText := fmt.Sprintf(`SELECT vin, ts, longitude, latitude
FROM %s.vehicle_locations
WHERE %s
ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
	rows, err := tdDB.QueryContext(ctx, sqlText)
	if err != nil {
		return 0, err
	}
	defer rows.Close()
	states := map[string]*gpsCoordinateBackfillState{}
	for rows.Next() {
		var vin string
		var eventTime time.Time
		var longitude, latitude float64
		if err := rows.Scan(&vin, &eventTime, &longitude, &latitude); err != nil {
			return 0, err
		}
		vin = strings.TrimSpace(vin)
		eventTime = eventTime.In(cfg.Location)
		date := eventTime.Format("2006-01-02")
		target := vin + "|" + date
		if _, ok := existing[target]; ok {
			continue
		}
		current := stats.GPSMileagePoint{
			VIN:       vin,
			StatDate:  date,
			Protocol:  envelope.ProtocolJT808,
			EventTime: eventTime,
			Longitude: longitude,
			Latitude:  latitude,
		}
		state := states[target]
		if state == nil {
			states[target] = &gpsCoordinateBackfillState{
				previous:        current,
				firstEventTime:  eventTime,
				latestEventTime: eventTime,
				pointCount:      1,
			}
			continue
		}
		segment := stats.CalculateGPSMileageSegment(state.previous, current)
		if segment.OutOfOrder {
			continue
		}
		state.previous = current
		state.latestEventTime = eventTime
		state.pointCount++
		if segment.Usable {
			state.distanceKM += segment.DistanceKM
			state.usableSegments++
		}
		if segment.BadJump {
			state.badJumps++
		}
		if segment.LongGap {
			state.longGaps++
		}
	}
	if err := rows.Err(); err != nil {
		return 0, err
	}
	added := 0
	for target, state := range states {
		if state.usableSegments == 0 {
			continue
		}
		parts := strings.SplitN(target, "|", 2)
		if len(parts) != 2 {
			continue
		}
		vin, date := parts[0], parts[1]
		sourceKey := string(envelope.ProtocolJT808) + ":" + vin + "@GPS_COORDINATE"
		key := vin + "|" + date + "|" + string(envelope.ProtocolJT808) + "|" + sourceKey
		aggregates[key] = &metricAgg{
			VIN:             vin,
			Date:            date,
			Protocol:        envelope.ProtocolJT808,
			FirstKM:         0,
			LatestKM:        state.distanceKM,
			Count:           state.pointCount,
			SourceKey:       sourceKey,
			SourceEndpoint:  "gps-coordinate",
			PlatformName:    "JT808 GPS轨迹估算",
			SourceKind:      "UNKNOWN",
			FirstEventTime:  state.firstEventTime,
			LatestEventTime: state.latestEventTime,
			QualityStatus:   stats.QualityOK,
			QualityReason:   stats.QualityReasonGPSCoordinate,
		}
		added++
	}
	if len(states) > 0 {
		slog.Info("JT808 GPS-coordinate fallbacks loaded", "states", len(states), "added", added)
	}
	return added, nil
}

func existingJT808OdometerTargets(ctx context.Context, db *sql.DB, cfg config) (map[string]struct{}, error) {
	result := map[string]struct{}{}
	if db == nil {
		return result, nil
	}
	rows, err := db.QueryContext(ctx, `SELECT DISTINCT vin, DATE_FORMAT(stat_date, '%Y-%m-%d')
FROM vehicle_daily_mileage_source
WHERE protocol = 'JT808'
  AND stat_date >= ? AND stat_date <= ?
  AND quality_status = ?
  AND COALESCE(quality_reason, '') <> ?
  AND latest_total_mileage_km IS NOT NULL`, cfg.DateFrom, cfg.DateTo, stats.QualityOK, stats.QualityReasonGPSCoordinate)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var vin, date string
		if err := rows.Scan(&vin, &date); err != nil {
			return nil, err
		}
		result[strings.TrimSpace(vin)+"|"+strings.TrimSpace(date)] = struct{}{}
	}
	return result, rows.Err()
}

func containsProtocol(protocols []envelope.Protocol, target envelope.Protocol) bool {
	for _, protocol := range protocols {
		if protocol == target {
			return true
		}
	}
	return false
}

func indexAggregateSourceRowsByDate(aggregates map[string]*metricAgg, protocol envelope.Protocol) map[string]map[string][]dailySourceLast {
	indexed := map[string]map[string][]dailySourceLast{}
	for _, agg := range aggregates {
		if agg == nil || agg.Protocol != protocol || strings.TrimSpace(agg.Date) == "" {
			continue
		}
		rows := indexed[agg.Date]
		if rows == nil {
			rows = map[string][]dailySourceLast{}
			indexed[agg.Date] = rows
		}
		rows[agg.VIN] = append(rows[agg.VIN], dailySourceLast{
			VIN:            agg.VIN,
			SourceKey:      agg.SourceKey,
			Phone:          agg.Phone,
			DeviceID:       agg.DeviceID,
			SourceEndpoint: agg.SourceEndpoint,
			SourceCode:     agg.SourceCode,
			PlatformName:   agg.PlatformName,
			SourceKind:     agg.SourceKind,
			FirstTS:        agg.LatestEventTime,
			TS:             agg.LatestEventTime,
			FirstTotalKM:   agg.LatestKM,
			TotalKM:        agg.LatestKM,
			RawSampleCount: agg.Count,
		})
	}
	return indexed
}

func rememberLatestSourceRows(history map[sourceHistoryID]dailySourceLast, rows map[string][]dailySourceLast) {
	for vin, sourceRows := range rows {
		for _, row := range sourceRows {
			rowVIN := strings.TrimSpace(row.VIN)
			if rowVIN == "" {
				rowVIN = strings.TrimSpace(vin)
				row.VIN = rowVIN
			}
			id := sourceHistoryID{VIN: rowVIN, SourceKey: strings.TrimSpace(row.SourceKey)}
			if id.VIN == "" || id.SourceKey == "" || row.TS.IsZero() {
				continue
			}
			if existing, ok := history[id]; !ok || row.TS.After(existing.TS) {
				history[id] = row
			}
		}
	}
}

func latestHistoricalSourceRow(history map[sourceHistoryID]dailySourceLast, current dailySourceLast) (dailySourceLast, bool) {
	id := sourceHistoryID{
		VIN:       strings.TrimSpace(current.VIN),
		SourceKey: strings.TrimSpace(current.SourceKey),
	}
	previous, ok := history[id]
	return previous, ok && previous.TS.Before(current.TS)
}

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
	}
	return queryDurablePreviousSourceRow(ctx, db, date, protocol, current)
}

func aggregateFromDailySource(date string, protocol envelope.Protocol, current dailySourceLast, previous dailySourceLast, hasPrevious bool) *metricAgg {
	firstKM := current.FirstTotalKM
	firstEventTime := current.FirstTS
	qualityStatus := stats.QualityOK
	qualityReason := stats.QualityReasonCurrentDayFirst
	if hasPrevious {
		firstKM = previous.TotalKM
		firstEventTime = previous.TS
		qualityStatus = stats.QualityOK
		qualityReason = stats.QualityReasonHistorical
	}
	if firstEventTime.IsZero() {
		firstEventTime = current.TS
	}
	count := current.RawSampleCount
	if count <= 0 {
		count = 1
	}
	agg := &metricAgg{
		VIN:             current.VIN,
		Date:            date,
		Protocol:        protocol,
		FirstKM:         firstKM,
		LatestKM:        current.TotalKM,
		Count:           count,
		SourceKey:       current.SourceKey,
		Phone:           current.Phone,
		DeviceID:        current.DeviceID,
		SourceEndpoint:  current.SourceEndpoint,
		SourceCode:      current.SourceCode,
		PlatformName:    current.PlatformName,
		SourceKind:      current.SourceKind,
		FirstEventTime:  firstEventTime,
		LatestEventTime: current.TS,
		QualityStatus:   qualityStatus,
		QualityReason:   qualityReason,
	}
	candidate := stats.SourceMileageSample{
		FirstTotalKM:    agg.FirstKM,
		LatestTotalKM:   agg.LatestKM,
		DailyKM:         stats.DailyMileageFromDayBoundary(agg.FirstKM, agg.LatestKM),
		FirstEventTime:  agg.FirstEventTime,
		LatestEventTime: agg.LatestEventTime,
		QualityStatus:   agg.QualityStatus,
		QualityReason:   agg.QualityReason,
	}
	stats.ApplyMileageQualityRules(&candidate)
	agg.QualityStatus = candidate.QualityStatus
	agg.QualityReason = candidate.QualityReason
	return agg
}

func queryDurablePreviousSourceRow(ctx context.Context, db *sql.DB, date string, protocol envelope.Protocol, current dailySourceLast) (dailySourceLast, bool, error) {
	if db == nil {
		return dailySourceLast{}, false, nil
	}
	totalKM, eventTime, found, err := stats.LookupLatestSourceBaselineBefore(ctx, db, current.VIN, date, protocol, current.SourceKey)
	if err != nil || !found {
		return dailySourceLast{}, found, err
	}
	previous := current
	previous.FirstTS = eventTime
	previous.TS = eventTime
	previous.FirstTotalKM = totalKM
	previous.TotalKM = totalKM
	return previous, true, nil
}

type trustedChoice struct {
	current  dailySourceLast
	previous dailySourceLast
}

func chooseTrustedSource(current []dailySourceLast, previous []dailySourceLast) (trustedChoice, bool) {
	previousBySource := map[string]dailySourceLast{}
	for _, row := range previous {
		previousBySource[row.SourceKey] = row
	}
	var chosen trustedChoice
	var chosenDelta float64
	for _, currentRow := range current {
		previousRow, ok := previousBySource[currentRow.SourceKey]
		if !ok {
			continue
		}
		delta, ok, _ := stats.NormalizeDailyMileageDeltaForWindow(stats.DailyMileageFromDayBoundary(previousRow.TotalKM, currentRow.TotalKM), previousRow.TS, currentRow.TS)
		if !ok {
			continue
		}
		if chosen.current.SourceKey == "" || delta < chosenDelta {
			chosen = trustedChoice{current: currentRow, previous: previousRow}
			chosenDelta = delta
		}
	}
	return chosen, chosen.current.SourceKey != ""
}

func queryDailyLastSourceRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string][]dailySourceLast, error) {
	where := backfillTimePredicates(cfg, date, nextDate(date))
	where = append(where,
		"parse_status = 'OK'",
		"vin IS NOT NULL",
		"vin <> ''",
		fmt.Sprintf("protocol = '%s'", quote(string(protocol))),
		realtimeMileageFramePredicate(),
	)
	if predicate := mileageBearingFramePredicate(protocol); predicate != "" {
		where = append(where, predicate)
	}
	sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint,
  FIRST(event_time), FIRST(parsed_json), FIRST(raw_text),
  LAST(event_time), LAST(parsed_json), LAST(raw_text), COUNT(*)
FROM %s.raw_frames
WHERE %s
GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
	return querySourceRows(ctx, db, cfg, protocol, sqlText, true)
}

func queryPreviousLastSourceRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string][]dailySourceLast, error) {
	// The baseline is the nearest earlier sample, not necessarily yesterday's.
	// LAST aggregates the complete pre-window history once per source so empty
	// calendar days do not make a backfill fall back to the current day's first
	// sample.
	where := backfillBeforePredicates(cfg, date)
	where = append(where,
		"parse_status = 'OK'",
		"vin IS NOT NULL",
		"vin <> ''",
		fmt.Sprintf("protocol = '%s'", quote(string(protocol))),
		realtimeMileageFramePredicate(),
	)
	if predicate := mileageBearingFramePredicate(protocol); predicate != "" {
		where = append(where, predicate)
	}
	sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint,
  LAST(event_time), LAST(parsed_json), LAST(raw_text), COUNT(*)
FROM %s.raw_frames
WHERE %s
GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
	return querySourceRows(ctx, db, cfg, protocol, sqlText, false)
}

func backfillBeforePredicates(cfg config, eventDateExclusive string) []string {
	lookbackDays := cfg.BaselineLookback
	if lookbackDays <= 0 {
		lookbackDays = 7
	}
	eventDateFrom := shiftDate(eventDateExclusive, -lookbackDays)
	where := []string{
		fmt.Sprintf("event_time >= '%s 00:00:00'", quote(eventDateFrom)),
		fmt.Sprintf("event_time < '%s 00:00:00'", quote(eventDateExclusive)),
	}
	if !cfg.EventTimeFullScan {
		where = append([]string{
			fmt.Sprintf("ts >= '%s 00:00:00'", quote(previousDate(eventDateFrom))),
			fmt.Sprintf("ts < '%s 00:00:00'", quote(nextDate(eventDateExclusive))),
		}, where...)
	}
	return where
}

func queryRealtimeLocationLastRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string][]dailySourceLast, error) {
	if db == nil {
		return nil, nil
	}
	loc := cfg.Location
	if loc == nil {
		loc = time.FixedZone("Asia/Shanghai", 8*3600)
	}
	eventTimeColumn := "l.total_mileage_event_time"
	timePredicate := `l.total_mileage_event_time >= ?
  AND l.total_mileage_event_time < DATE_ADD(?, INTERVAL 1 DAY)`
	if cfg.StationaryCarry {
		eventTimeColumn = "l.event_time"
		timePredicate = `l.event_time >= ?
  AND l.event_time < DATE_ADD(?, INTERVAL 1 DAY)
  AND l.total_mileage_event_time < ?
  AND COALESCE(l.speed_kmh, 0) = 0`
	}
	sqlText := `SELECT l.vin, COALESCE(NULLIF(s.peer, ''), ''), ` + eventTimeColumn + `, l.total_mileage_km
FROM vehicle_realtime_location l
LEFT JOIN vehicle_realtime_snapshot s ON s.protocol = l.protocol AND s.vin = l.vin
WHERE l.protocol = ?
  AND l.vin IS NOT NULL AND l.vin <> ''
  AND l.total_mileage_km IS NOT NULL AND l.total_mileage_km > 0
  AND ` + timePredicate
	args := []any{string(protocol), date, date}
	if cfg.StationaryCarry {
		args = append(args, date)
	}
	rows, err := db.QueryContext(ctx, sqlText, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[string][]dailySourceLast{}
	for rows.Next() {
		var vin string
		var peer sql.NullString
		var eventTime time.Time
		var totalKM float64
		if err := rows.Scan(&vin, &peer, &eventTime, &totalKM); err != nil {
			return nil, err
		}
		vin = strings.TrimSpace(vin)
		if vin == "" || totalKM <= 0 {
			continue
		}
		sourceEndpoint := strings.TrimSpace(peer.String)
		if sourceEndpoint == "" && protocol == envelope.ProtocolYutongMQTT {
			sourceEndpoint = "mqtt://yutong/realtime-location"
		}
		deviceID := ""
		if protocol == envelope.ProtocolYutongMQTT {
			deviceID = vin
		}
		sourceCode, platformName, sourceKind := knownPlatformSourceMetadata(protocol, sourceEndpoint)
		sourceKey := normalizedSourceKeyForSource(string(protocol), "", deviceID, sourceEndpoint, sourceKind, sourceCode)
		if sourceKey == "" {
			continue
		}
		row := dailySourceLast{
			VIN:            vin,
			SourceKey:      sourceKey,
			DeviceID:       deviceID,
			SourceEndpoint: sourceEndpoint,
			SourceCode:     sourceCode,
			PlatformName:   platformName,
			SourceKind:     sourceKind,
			FirstTS:        eventTime.In(loc),
			TS:             eventTime.In(loc),
			FirstTotalKM:   totalKM,
			TotalKM:        totalKM,
			RawSampleCount: 1,
		}
		out[vin] = append(out[vin], row)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

func querySourceRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, sqlText string, includeFirst bool) (map[string][]dailySourceLast, error) {
	rows, err := db.QueryContext(ctx, sqlText)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	latestBySource := map[string]dailySourceLast{}
	for rows.Next() {
		var vin string
		var phone string
		var deviceID string
		var sourceEndpoint string
		var firstTS time.Time
		var firstParsedJSON string
		var firstRawText string
		var ts time.Time
		var parsedJSON string
		var rawText string
		var rawSampleCount int64
		if includeFirst {
			if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &firstTS, &firstParsedJSON, &firstRawText, &ts, &parsedJSON, &rawText, &rawSampleCount); err != nil {
				return nil, err
			}
		} else {
			if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &ts, &parsedJSON, &rawText, &rawSampleCount); err != nil {
				return nil, err
			}
			firstTS = ts
			firstParsedJSON = parsedJSON
			firstRawText = rawText
		}
		latestTotalKM, ok := mileageFromEvidence(protocol, vin, parsedJSON, rawText, ts, cfg.Location)
		if !ok {
			continue
		}
		firstTotalKM := latestTotalKM
		if includeFirst {
			if parsedFirst, ok := mileageFromEvidence(protocol, vin, firstParsedJSON, firstRawText, firstTS, cfg.Location); ok {
				firstTotalKM = parsedFirst
			}
		}
		sourceCode, platformName, sourceKind := knownPlatformSourceMetadata(protocol, sourceEndpoint)
		sourceKey := normalizedSourceKeyForSource(string(protocol), phone, deviceID, sourceEndpoint, sourceKind, sourceCode)
		if sourceKey == "" {
			continue
		}
		row := dailySourceLast{
			VIN:            strings.TrimSpace(vin),
			SourceKey:      sourceKey,
			Phone:          strings.TrimSpace(phone),
			DeviceID:       strings.TrimSpace(deviceID),
			SourceEndpoint: strings.TrimSpace(sourceEndpoint),
			SourceCode:     sourceCode,
			PlatformName:   platformName,
			SourceKind:     sourceKind,
			FirstTS:        firstTS.In(cfg.Location),
			TS:             ts.In(cfg.Location),
			FirstTotalKM:   firstTotalKM,
			TotalKM:        latestTotalKM,
			RawSampleCount: rawSampleCount,
		}
		key := row.VIN + "|" + row.SourceKey
		if existing, ok := latestBySource[key]; !ok || row.TS.After(existing.TS) {
			latestBySource[key] = row
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	out := map[string][]dailySourceLast{}
	for _, row := range latestBySource {
		out[row.VIN] = append(out[row.VIN], row)
	}
	return out, nil
}

func mileageFromParsed(protocol envelope.Protocol, vin string, parsedJSON string, eventTime time.Time, loc *time.Location) (float64, bool) {
	fields := fieldsForStats(protocol, vin, parsedJSON)
	env := envelope.FrameEnvelope{
		Protocol:     protocol,
		VIN:          strings.TrimSpace(vin),
		EventTimeMS:  eventTime.UnixMilli(),
		ReceivedAtMS: eventTime.UnixMilli(),
		Fields:       fields,
		ParsedFields: fields,
		ParseStatus:  envelope.ParseOK,
	}
	samples, err := stats.SamplesFromEnvelope(env, loc)
	if err != nil || len(samples) == 0 {
		return 0, false
	}
	return samples[0].TotalMileageKM, true
}

func mileageFromEvidence(protocol envelope.Protocol, vin string, parsedJSON string, rawText string, eventTime time.Time, loc *time.Location) (float64, bool) {
	if totalKM, ok := mileageFromParsed(protocol, vin, parsedJSON, eventTime, loc); ok {
		return totalKM, true
	}
	if protocol != envelope.ProtocolYutongMQTT || strings.TrimSpace(rawText) == "" {
		return 0, false
	}
	return mileageFromParsed(protocol, vin, rawText, eventTime, loc)
}

func normalizedSourceKey(protocol string, phone string, deviceID string, endpoint string) string {
	return normalizedSourceKeyForSource(protocol, phone, deviceID, endpoint, "", "")
}

func normalizedSourceKeyForSource(protocol string, phone string, deviceID string, endpoint string, sourceKind string, sourceCode string) string {
	sourceIP := stats.NormalizeSourceIP(endpoint)
	return stats.SourceKeyForSource(envelope.Protocol(protocol), phone, deviceID, sourceIP, sourceKind, sourceCode)
}

func knownPlatformSourceMetadata(protocol envelope.Protocol, endpoint string) (sourceCode string, platformName string, sourceKind string) {
	if protocol == envelope.ProtocolYutongMQTT && stats.NormalizeSourceIP(endpoint) == "mqtt" {
		return "yutong", "宇通", "PLATFORM"
	}
	return "", "", ""
}

func fieldsForStats(protocol envelope.Protocol, vin string, text string) map[string]any {
	fields := map[string]any{}
	text = strings.TrimSpace(text)
	if text == "" {
		return fields
	}
	parsed := map[string]any{}
	if err := json.Unmarshal([]byte(text), &parsed); err == nil && len(parsed) > 0 {
		if flattened, _, ok := realtime.ComputeParsedFieldsForEnvelope(envelope.FrameEnvelope{
			Protocol: protocol,
			VIN:      vin,
			Parsed:   parsed,
		}); ok {
			fields = flattened
		} else {
			fields = parsed
		}
	}
	for _, key := range mileageKeys(protocol) {
		if _, exists := fields[key]; exists {
			continue
		}
		value, ok := extractJSONStringField(text, key)
		if !ok {
			continue
		}
		fields[key] = value
	}
	return fields
}

func mileageKeys(protocol envelope.Protocol) []string {
	switch protocol {
	case envelope.ProtocolGB32960:
		return []string{"gb32960.vehicle.total_mileage_km"}
	case envelope.ProtocolJT808:
		return []string{"jt808.location.total_mileage_km"}
	case envelope.ProtocolYutongMQTT:
		return []string{
			"yutong_mqtt.data.total_mileage_km",
			"yutong_mqtt.root.data.total_mileage_km",
			"yutong_mqtt.data.total_mileage",
			"yutong_mqtt.root.data.total_mileage",
		}
	default:
		return nil
	}
}

func extractJSONStringField(text string, key string) (string, bool) {
	pattern := regexp.QuoteMeta(`"`+key+`"`) + `\s*:\s*"?([^",}]+)"?`
	match := regexp.MustCompile(pattern).FindStringSubmatch(text)
	if len(match) != 2 {
		return "", false
	}
	return strings.TrimSpace(match[1]), true
}

func loadConfig() (config, error) {
	loc, err := time.LoadLocation(env("LOCAL_TZ", "Asia/Shanghai"))
	if err != nil {
		loc = time.FixedZone("Asia/Shanghai", 8*3600)
	}
	dateFrom, dateTo := resolveBackfillDateRange(time.Now(), loc)
	protocols, err := parseProtocols(env("BACKFILL_PROTOCOLS", "GB32960,JT808,YUTONG_MQTT"))
	if err != nil {
		return config{}, err
	}
	return config{
		MySQLDSN:          env("MYSQL_DSN", ""),
		TDengineDriver:    env("TDENGINE_DRIVER", "taosWS"),
		TDengineDSN:       env("TDENGINE_DSN", ""),
		TDengineDatabase:  env("TDENGINE_DATABASE", history.DefaultDatabase),
		DateFrom:          dateFrom,
		DateTo:            dateTo,
		Protocols:         protocols,
		Method:            env("BACKFILL_METHOD", "last_diff"),
		Limit:             envInt("BACKFILL_LIMIT", 0),
		DryRun:            envBool("BACKFILL_DRY_RUN", true),
		Reset:             envBool("BACKFILL_RESET", false),
		Debug:             envBool("BACKFILL_DEBUG", false),
		EventTimeFullScan: envBool("BACKFILL_EVENT_TIME_FULL_SCAN", false),
		StationaryCarry:   envBool("BACKFILL_STATIONARY_CARRY_FORWARD", false),
		JT808GPSFallback:  envBool("BACKFILL_JT808_GPS_FALLBACK", true),
		ProgressEvery:     int64(envInt("BACKFILL_PROGRESS_EVERY", 100000)),
		BaselineLookback:  envInt("BACKFILL_BASELINE_LOOKBACK_DAYS", 7),
		Location:          loc,
	}, nil
}

func resolveBackfillDateRange(now time.Time, loc *time.Location) (string, string) {
	if loc == nil {
		loc = time.FixedZone("Asia/Shanghai", 8*3600)
	}
	explicitFrom := strings.TrimSpace(os.Getenv("BACKFILL_DATE_FROM"))
	explicitTo := strings.TrimSpace(os.Getenv("BACKFILL_DATE_TO"))
	if explicitTo != "" || explicitFrom != "" {
		dateTo := explicitTo
		if dateTo == "" {
			dateTo = now.In(loc).Format("2006-01-02")
		}
		dateFrom := explicitFrom
		if dateFrom == "" {
			dateFrom = dateTo
		}
		return dateFrom, dateTo
	}
	daysBack := envInt("BACKFILL_DAYS_BACK", 0)
	if daysBack < 0 {
		daysBack = 0
	}
	windowDays := envInt("BACKFILL_WINDOW_DAYS", 1)
	if windowDays < 1 {
		windowDays = 1
	}
	dateToTime := now.In(loc).AddDate(0, 0, -daysBack)
	dateFromTime := dateToTime.AddDate(0, 0, -(windowDays - 1))
	return dateFromTime.Format("2006-01-02"), dateToTime.Format("2006-01-02")
}

func queryRawFrames(ctx context.Context, db *sql.DB, cfg config) (*sql.Rows, error) {
	where := backfillTimePredicates(cfg, cfg.DateFrom, nextDate(cfg.DateTo))
	where = append(where,
		"parse_status = 'OK'",
		"vin IS NOT NULL",
		"vin <> ''",
	)
	if len(cfg.Protocols) > 0 {
		quoted := make([]string, 0, len(cfg.Protocols))
		for _, protocol := range cfg.Protocols {
			quoted = append(quoted, "'"+quote(string(protocol))+"'")
		}
		where = append(where, "protocol IN ("+strings.Join(quoted, ",")+")")
	}
	where = append(where, realtimeMileageFramePredicate())
	sqlText := fmt.Sprintf(`SELECT protocol, vin, phone, device_id, source_endpoint, event_id, message_id, event_time, received_at, parsed_json, raw_text
FROM %s.raw_frames
WHERE %s
	ORDER BY event_time ASC, ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
	if cfg.Limit > 0 {
		sqlText += fmt.Sprintf(" LIMIT %d", cfg.Limit)
	}
	slog.Info("query raw frames", "date_from", cfg.DateFrom, "date_to", cfg.DateTo, "protocols", cfg.Protocols, "limit", cfg.Limit)
	return db.QueryContext(ctx, sqlText)
}

func backfillTimePredicates(cfg config, eventDateFrom string, eventDateToExclusive string) []string {
	where := make([]string, 0, 4)
	if !cfg.EventTimeFullScan {
		// ts is the TDengine primary timestamp and may be adjusted slightly to keep
		// rows unique. Use it only as a broad index-friendly window; event_time is
		// the protocol business boundary used for the final natural-day result.
		where = append(where,
			fmt.Sprintf("ts >= '%s 00:00:00'", quote(previousDate(eventDateFrom))),
			fmt.Sprintf("ts < '%s 00:00:00'", quote(nextDate(eventDateToExclusive))),
		)
	}
	return append(where,
		fmt.Sprintf("event_time >= '%s 00:00:00'", quote(eventDateFrom)),
		fmt.Sprintf("event_time < '%s 00:00:00'", quote(eventDateToExclusive)),
	)
}

func realtimeMileageFramePredicate() string {
	return `(
  (protocol = 'GB32960' AND message_id IN (2,3))
  OR (protocol = 'JT808' AND message_id = 512)
  OR (protocol = 'YUTONG_MQTT')
)`
}

func mileageBearingFramePredicate(protocol envelope.Protocol) string {
	tokens := mileageSearchTokens(protocol)
	if len(tokens) == 0 {
		return ""
	}
	conditions := make([]string, 0, len(tokens))
	for _, token := range tokens {
		conditions = append(conditions, fmt.Sprintf("parsed_json LIKE '%%%s%%'", quote(token)))
		if protocol == envelope.ProtocolYutongMQTT {
			conditions = append(conditions, fmt.Sprintf("raw_text LIKE '%%%s%%'", quote(token)))
		}
	}
	return "(" + strings.Join(conditions, " OR ") + ")"
}

func mileageSearchTokens(protocol envelope.Protocol) []string {
	tokens := append([]string{}, mileageKeys(protocol)...)
	switch protocol {
	case envelope.ProtocolYutongMQTT:
		tokens = append(tokens, "TOTAL_MILEAGE", "totalMileage", "total_mileage_km")
	}
	seen := make(map[string]struct{}, len(tokens))
	out := make([]string, 0, len(tokens))
	for _, token := range tokens {
		token = strings.TrimSpace(token)
		if token == "" {
			continue
		}
		if _, ok := seen[token]; ok {
			continue
		}
		seen[token] = struct{}{}
		out = append(out, token)
	}
	return out
}

func scanRawFrame(rows *sql.Rows) (rawFrameRow, error) {
	var protocol, vin, phone, deviceID, sourceEndpoint, eventID, parsed, rawText string
	var messageID int64
	var eventTime, receivedAt time.Time
	if err := rows.Scan(&protocol, &vin, &phone, &deviceID, &sourceEndpoint, &eventID, &messageID, &eventTime, &receivedAt, &parsed, &rawText); err != nil {
		return rawFrameRow{}, err
	}
	return rawFrameRow{
		Protocol:       envelope.Protocol(strings.TrimSpace(protocol)),
		VIN:            strings.TrimSpace(vin),
		Phone:          strings.TrimSpace(phone),
		DeviceID:       strings.TrimSpace(deviceID),
		SourceEndpoint: strings.TrimSpace(sourceEndpoint),
		EventID:        strings.TrimSpace(eventID),
		MessageID:      fmt.Sprintf("0x%04X", messageID),
		EventTimeMS:    eventTime.UnixMilli(),
		ReceivedAtMS:   receivedAt.UnixMilli(),
		ParsedJSON:     parsed,
		RawText:        rawText,
	}, nil
}

func isChunkManifest(text string) bool {
	if !strings.Contains(text, "chunked") || !strings.Contains(text, "payload_kind") {
		return false
	}
	var manifest struct {
		Chunked bool `json:"chunked"`
	}
	return json.Unmarshal([]byte(text), &manifest) == nil && manifest.Chunked
}

func readChunkedPayload(ctx context.Context, db *sql.DB, database string, eventID string, kind string) (string, error) {
	sqlText := fmt.Sprintf(`SELECT chunk_text FROM %s.raw_frame_payload_chunks
WHERE event_id = '%s' AND payload_kind = '%s'
ORDER BY chunk_index ASC`, ident(database), quote(eventID), quote(kind))
	rows, err := db.QueryContext(ctx, sqlText)
	if err != nil {
		return "", err
	}
	defer rows.Close()
	var b strings.Builder
	for rows.Next() {
		var chunk string
		if err := rows.Scan(&chunk); err != nil {
			return "", err
		}
		b.WriteString(chunk)
	}
	if err := rows.Err(); err != nil {
		return "", err
	}
	if b.Len() == 0 {
		return "", fmt.Errorf("no chunks found")
	}
	return b.String(), nil
}

func resetStats(ctx context.Context, db *sql.DB, cfg config) (int64, error) {
	clauses := []string{"stat_date >= ?", "stat_date <= ?"}
	args := []any{cfg.DateFrom, cfg.DateTo}
	if len(cfg.Protocols) > 0 {
		placeholders := make([]string, 0, len(cfg.Protocols))
		for _, protocol := range cfg.Protocols {
			placeholders = append(placeholders, "?")
			args = append(args, string(protocol))
		}
		clauses = append(clauses, "protocol IN ("+strings.Join(placeholders, ",")+")")
	}
	var deleted int64
	for _, table := range []string{"vehicle_daily_mileage_source", "vehicle_daily_mileage"} {
		result, err := db.ExecContext(ctx, "DELETE FROM "+table+" WHERE "+strings.Join(clauses, " AND "), args...)
		if err != nil {
			return deleted, err
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return deleted, err
		}
		deleted += affected
	}
	return deleted, nil
}

func parseProtocols(value string) ([]envelope.Protocol, error) {
	var protocols []envelope.Protocol
	for _, part := range strings.Split(value, ",") {
		part = strings.ToUpper(strings.TrimSpace(part))
		if part == "" {
			continue
		}
		switch envelope.Protocol(part) {
		case envelope.ProtocolGB32960, envelope.ProtocolJT808, envelope.ProtocolYutongMQTT:
			protocols = append(protocols, envelope.Protocol(part))
		default:
			return nil, fmt.Errorf("unsupported protocol %q", part)
		}
	}
	return protocols, nil
}

func loadEnvFiles(paths string) error {
	for _, path := range strings.Split(paths, ",") {
		if err := loadEnvFile(path); err != nil {
			return err
		}
	}
	return nil
}

func backfillEnvFiles(explicitFiles string, explicitFile string, defaults []string) string {
	if value := strings.TrimSpace(explicitFiles); value != "" {
		return value
	}
	if value := strings.TrimSpace(explicitFile); value != "" {
		return value
	}
	var existing []string
	for _, path := range defaults {
		path = strings.TrimSpace(path)
		if path == "" {
			continue
		}
		if _, err := os.Stat(path); err == nil {
			existing = append(existing, path)
		}
	}
	return strings.Join(existing, ",")
}

func loadEnvFile(path string) error {
	path = strings.TrimSpace(path)
	if path == "" {
		return nil
	}
	data, err := os.ReadFile(path)
	if err != nil {
		return err
	}
	for _, line := range strings.Split(string(data), "\n") {
		line = strings.TrimSpace(line)
		if line == "" || strings.HasPrefix(line, "#") || !strings.Contains(line, "=") {
			continue
		}
		key, value, _ := strings.Cut(line, "=")
		key = strings.TrimSpace(key)
		value = strings.Trim(strings.TrimSpace(value), "\"'")
		if key != "" && os.Getenv(key) == "" {
			_ = os.Setenv(key, value)
		}
	}
	return nil
}

func env(key string, fallback string) string {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback
	}
	return value
}

func envInt(key string, fallback int) int {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback
	}
	parsed, err := strconv.Atoi(value)
	if err != nil {
		return fallback
	}
	return parsed
}

func envBool(key string, fallback bool) bool {
	value := strings.ToLower(strings.TrimSpace(os.Getenv(key)))
	if value == "" {
		return fallback
	}
	return value == "1" || value == "true" || value == "yes" || value == "y"
}

func nextDate(date string) string {
	parsed, err := time.Parse("2006-01-02", date)
	if err != nil {
		return date
	}
	return parsed.AddDate(0, 0, 1).Format("2006-01-02")
}

func previousDate(date string) string {
	parsed, err := time.Parse("2006-01-02", date)
	if err != nil {
		return date
	}
	return parsed.AddDate(0, 0, -1).Format("2006-01-02")
}

func shiftDate(date string, days int) string {
	parsed, err := time.Parse("2006-01-02", date)
	if err != nil {
		return date
	}
	return parsed.AddDate(0, 0, days).Format("2006-01-02")
}

func dateRangeWithPrevious(from string, to string) ([]string, error) {
	dates, err := dateRange(from, to)
	if err != nil {
		return nil, err
	}
	return append([]string{previousDate(from)}, dates...), nil
}

func dateRange(from string, to string) ([]string, error) {
	start, err := time.Parse("2006-01-02", from)
	if err != nil {
		return nil, err
	}
	end, err := time.Parse("2006-01-02", to)
	if err != nil {
		return nil, err
	}
	if end.Before(start) {
		return nil, fmt.Errorf("dateTo before dateFrom")
	}
	var dates []string
	for day := start; !day.After(end); day = day.AddDate(0, 0, 1) {
		dates = append(dates, day.Format("2006-01-02"))
	}
	return dates, nil
}

func ident(value string) string {
	value = strings.TrimSpace(strings.Trim(value, "`"))
	if value == "" {
		return history.DefaultDatabase
	}
	return value
}

func quote(value string) string {
	return strings.ReplaceAll(value, "'", "''")
}

func fail(stage string, err error) {
	slog.Error(stage, "error", err)
	os.Exit(1)
}