lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats/daily_metric.go

package stats

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

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

const (
	maxNegativeMileageJitterKM     = 1.0
	defaultCacheRetention          = 72 * time.Hour
	defaultCacheCleanupInterval    = 10 * time.Minute
	defaultBaselineMissTTL         = time.Minute
	defaultBaselineHitTTL          = 5 * time.Minute
	defaultMaxCacheEntries         = 1000000
	defaultGPSAccumulationInterval = 10 * time.Second
)

type Execer interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
}

type Writer struct {
	exec                    Execer
	query                   Queryer
	loc                     *time.Location
	sourceTouchInterval     time.Duration
	projectionInterval      time.Duration
	cacheRetention          time.Duration
	cacheCleanupInterval    time.Duration
	baselineMissTTL         time.Duration
	baselineHitTTL          time.Duration
	maxCacheEntries         int
	lastCacheCleanup        time.Time
	lastCacheCleanupStats   cacheCleanupStats
	cacheEvictions          cacheCleanupStats
	mu                      sync.Mutex
	lastTotalMileage        map[string]float64
	lastSourceSeen          map[string]time.Time
	lastProjection          map[string]projectionCacheEntry
	pendingProjection       map[string]pendingProjectionEntry
	baselineCache           map[string]sourceBaselineCacheEntry
	mileageKeysByPrefix     map[string]map[string]struct{}
	projectionKeysByPrefix  map[string]map[string]struct{}
	baselineKeysByPrefix    map[string]map[string]struct{}
	lastGPSAccumulation     map[string]time.Time
	gpsAccumulationInterval time.Duration
}

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

type AppendResult struct {
	SamplesFound                     int
	SamplesWritten                   int
	SamplesSkippedMissingFields      int
	SamplesSkippedMissingVIN         int
	SamplesSkippedMissingMileage     int
	SamplesRecoveredStationary       int
	SamplesRecoveredGPSCoordinate    int
	SamplesSkippedNonMileageFrame    int
	SamplesSkippedNonPositiveMileage int
	SamplesSkippedMissingTime        int
	SamplesSkippedSameMileage        int
	SamplesSkippedMissingSource      int
	SamplesAdjustedFutureEventTime   int
	SourceTouchesAttempted           int
	SourceTouchesWritten             int
	SourceTouchesSkippedThrottled    int
	SourceTouchesSkippedMissing      int
	SourceTouchesSkippedUnmanaged    int
	ProjectionsAttempted             int
	ProjectionsWritten               int
	ProjectionsSkippedThrottled      int
}

type ProjectionFlushResult struct {
	Attempted int
	Written   int
	Failed    int
}

type ProjectionReconcileResult struct {
	Targets   int
	Attempted int
	Written   int
	Failed    int
}

type CacheStats struct {
	LastTotalMileageEntries  int
	LastSourceSeenEntries    int
	LastProjectionEntries    int
	PendingProjectionEntries int
	BaselineEntries          int
	GPSAccumulationEntries   int
	MaxEntries               int
	LastCleanupAt            time.Time
	LastCleanupTotalMileage  int
	LastCleanupSourceSeen    int
	LastCleanupProjection    int
	LastCleanupBaseline      int
	LastCleanupGPS           int
	TotalMileageEvictions    int
	SourceSeenEvictions      int
	ProjectionEvictions      int
	BaselineEvictions        int
	GPSEvictions             int
}

type cacheCleanupStats struct {
	totalMileage int
	sourceSeen   int
	projection   int
	baseline     int
	gps          int
}

func NewWriter(exec Execer, loc *time.Location) *Writer {
	if exec == nil {
		panic("stats execer must not be nil")
	}
	if loc == nil {
		loc = time.FixedZone("Asia/Shanghai", 8*3600)
	}
	writer := &Writer{
		exec:                    exec,
		loc:                     loc,
		sourceTouchInterval:     time.Minute,
		projectionInterval:      15 * time.Second,
		cacheRetention:          defaultCacheRetention,
		cacheCleanupInterval:    defaultCacheCleanupInterval,
		baselineMissTTL:         defaultBaselineMissTTL,
		baselineHitTTL:          defaultBaselineHitTTL,
		maxCacheEntries:         defaultMaxCacheEntries,
		lastTotalMileage:        map[string]float64{},
		lastSourceSeen:          map[string]time.Time{},
		lastProjection:          map[string]projectionCacheEntry{},
		pendingProjection:       map[string]pendingProjectionEntry{},
		baselineCache:           map[string]sourceBaselineCacheEntry{},
		mileageKeysByPrefix:     map[string]map[string]struct{}{},
		projectionKeysByPrefix:  map[string]map[string]struct{}{},
		baselineKeysByPrefix:    map[string]map[string]struct{}{},
		lastGPSAccumulation:     map[string]time.Time{},
		gpsAccumulationInterval: defaultGPSAccumulationInterval,
	}
	if query, ok := exec.(Queryer); ok {
		writer.query = query
	}
	return writer
}

func (w *Writer) SetSourceTouchInterval(interval time.Duration) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if interval < 0 {
		interval = 0
	}
	w.sourceTouchInterval = interval
}

func (w *Writer) SetProjectionInterval(interval time.Duration) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if interval < 0 {
		interval = 0
	}
	w.projectionInterval = interval
}

func (w *Writer) SetCacheRetention(retention time.Duration) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if retention < 0 {
		retention = 0
	}
	w.cacheRetention = retention
}

func (w *Writer) SetCacheCleanupInterval(interval time.Duration) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if interval < 0 {
		interval = 0
	}
	w.cacheCleanupInterval = interval
}

func (w *Writer) SetBaselineMissTTL(ttl time.Duration) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if ttl < 0 {
		ttl = 0
	}
	w.baselineMissTTL = ttl
}

func (w *Writer) SetBaselineHitTTL(ttl time.Duration) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if ttl < 0 {
		ttl = 0
	}
	w.baselineHitTTL = ttl
}

func (w *Writer) SetMaxCacheEntries(maxEntries int) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if maxEntries < 0 {
		maxEntries = 0
	}
	w.maxCacheEntries = maxEntries
	w.enforceCacheLimitsLocked()
}

func (w *Writer) SetGPSAccumulationInterval(interval time.Duration) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if interval < 0 {
		interval = 0
	}
	w.gpsAccumulationInterval = interval
}

func (w *Writer) CacheStats() CacheStats {
	w.mu.Lock()
	defer w.mu.Unlock()
	return CacheStats{
		LastTotalMileageEntries:  len(w.lastTotalMileage),
		LastSourceSeenEntries:    len(w.lastSourceSeen),
		LastProjectionEntries:    len(w.lastProjection),
		PendingProjectionEntries: len(w.pendingProjection),
		BaselineEntries:          len(w.baselineCache),
		GPSAccumulationEntries:   len(w.lastGPSAccumulation),
		MaxEntries:               w.maxCacheEntries,
		LastCleanupAt:            w.lastCacheCleanup,
		LastCleanupTotalMileage:  w.lastCacheCleanupStats.totalMileage,
		LastCleanupSourceSeen:    w.lastCacheCleanupStats.sourceSeen,
		LastCleanupProjection:    w.lastCacheCleanupStats.projection,
		LastCleanupBaseline:      w.lastCacheCleanupStats.baseline,
		LastCleanupGPS:           w.lastCacheCleanupStats.gps,
		TotalMileageEvictions:    w.cacheEvictions.totalMileage,
		SourceSeenEvictions:      w.cacheEvictions.sourceSeen,
		ProjectionEvictions:      w.cacheEvictions.projection,
		BaselineEvictions:        w.cacheEvictions.baseline,
		GPSEvictions:             w.cacheEvictions.gps,
	}
}

func (w *Writer) EnsureSchema(ctx context.Context) error {
	for _, statement := range []string{
		DataSourceTableSQL,
		GPSMileageStateTableSQL,
		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 && !isIgnorableSchemaChangeError(err) {
			return err
		}
	}
	return nil
}

func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
	_, err := w.AppendWithResult(ctx, env)
	return err
}

func (w *Writer) AppendWithResult(ctx context.Context, env envelope.FrameEnvelope) (AppendResult, error) {
	var result AppendResult
	if len(env.Fields) == 0 {
		result.SamplesSkippedMissingFields = 1
		return result, nil
	}
	seenAt := w.sourceSeenAt(env)
	w.maybeCleanupCaches(seenAt)
	identity, hasSource := NewSourceIdentityFromEnvelope(env)
	var sourceResult AppendResult
	if hasSource && ShouldManageDataSource(identity) {
		if w.shouldTouchSource(identity, seenAt) {
			sourceResult.SourceTouchesAttempted = 1
			if err := UpsertDataSource(ctx, w.exec, identity, seenAt); err != nil {
				return sourceResult, err
			}
			w.markSourceTouched(identity, seenAt)
			sourceResult.SourceTouchesWritten = 1
		} else {
			sourceResult.SourceTouchesSkippedThrottled = 1
		}
	} else if hasSource {
		sourceResult.SourceTouchesSkippedUnmanaged = 1
	}
	samples, extractionResult, err := samplesFromEnvelopeWithResult(env, w.loc)
	result = extractionResult
	result.SourceTouchesAttempted += sourceResult.SourceTouchesAttempted
	result.SourceTouchesWritten += sourceResult.SourceTouchesWritten
	result.SourceTouchesSkippedThrottled += sourceResult.SourceTouchesSkippedThrottled
	result.SourceTouchesSkippedMissing += sourceResult.SourceTouchesSkippedMissing
	result.SourceTouchesSkippedUnmanaged += sourceResult.SourceTouchesSkippedUnmanaged
	if err != nil {
		return result, err
	}
	var stationaryCandidate *SourceMileageSample
	if len(samples) == 0 && hasSource && result.SamplesSkippedMissingMileage > 0 {
		if point, ok := GPSMileagePointFromEnvelope(env, identity, w.loc); ok && w.shouldAccumulateGPS(point) {
			_, recovered, recoverErr := AccumulateGPSMileage(ctx, w.exec, point)
			if recoverErr != nil {
				return result, recoverErr
			}
			w.markGPSAccumulated(point)
			if recovered {
				result.SamplesFound++
				result.SamplesWritten++
				result.SamplesRecoveredGPSCoordinate++
				result.ProjectionsAttempted++
				result.ProjectionsWritten++
				return result, nil
			}
		}
		sample, candidate, recovered, recoverErr := w.stationaryCarryForward(ctx, env, identity)
		if recoverErr != nil {
			return result, recoverErr
		}
		if recovered {
			samples = []MetricSample{sample}
			stationaryCandidate = &candidate
			result.SamplesFound++
			result.SamplesRecoveredStationary++
		}
	}
	for _, sample := range samples {
		if !hasSource {
			result.SamplesSkippedMissingSource++
			result.SourceTouchesSkippedMissing++
			continue
		}
		if w.seenSameMileage(sample) {
			result.SamplesSkippedSameMileage++
			// A repeated cumulative odometer is usually a harmless duplicate,
			// but some upstream platforms keep reporting positions and speed
			// while their instrument mileage remains stale for hours. Preserve
			// the duplicate evidence and accumulate a speed-backed GPS
			// candidate so real movement does not disappear from daily mileage.
			if point, ok := GPSMileagePointFromEnvelope(env, identity, w.loc); ok && w.shouldAccumulateGPS(point) {
				_, recovered, recoverErr := AccumulateGPSMileage(ctx, w.exec, point)
				if recoverErr != nil {
					return result, recoverErr
				}
				w.markGPSAccumulated(point)
				if recovered {
					result.SamplesWritten++
					result.SamplesRecoveredGPSCoordinate++
					result.ProjectionsAttempted++
					result.ProjectionsWritten++
				}
			}
			continue
		}
		candidate := SourceMileageSampleFromMetric(sample, identity)
		if stationaryCandidate != nil {
			candidate = *stationaryCandidate
		} else {
			if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil {
				return result, err
			}
		}
		projectDaily := w.shouldProjectDailyMileage(sample)
		if projectDaily {
			result.ProjectionsAttempted++
		} else {
			result.ProjectionsSkippedThrottled++
		}
		if err := w.writeMileageSample(ctx, sample, candidate, projectDaily); err != nil {
			return result, err
		}
		w.markBaselineWritten(sample, candidate)
		if projectDaily {
			w.markProjected(sample)
			result.ProjectionsWritten++
		} else {
			w.markProjectionPending(sample, time.Now())
		}
		w.markMileageWritten(sample)
		result.SamplesWritten++
	}
	return result, nil
}

func (w *Writer) shouldAccumulateGPS(point GPSMileagePoint) bool {
	key := gpsMileageStateCacheKey(point)
	w.mu.Lock()
	defer w.mu.Unlock()
	last, ok := w.lastGPSAccumulation[key]
	if !ok || w.gpsAccumulationInterval == 0 {
		return true
	}
	return point.EventTime.After(last.Add(w.gpsAccumulationInterval))
}

func (w *Writer) markGPSAccumulated(point GPSMileagePoint) {
	key := gpsMileageStateCacheKey(point)
	w.mu.Lock()
	if previous, ok := w.lastGPSAccumulation[key]; !ok || point.EventTime.After(previous) {
		w.lastGPSAccumulation[key] = point.EventTime
	}
	w.enforceCacheLimitsLocked()
	w.mu.Unlock()
}

func gpsMileageStateCacheKey(point GPSMileagePoint) string {
	return point.VIN + "|" + point.StatDate + "|" + string(point.Protocol) + "|" + point.SourceKey
}

func (w *Writer) stationaryCarryForward(ctx context.Context, env envelope.FrameEnvelope, identity SourceIdentity) (MetricSample, SourceMileageSample, bool, error) {
	if env.Protocol != envelope.ProtocolYutongMQTT || strings.TrimSpace(env.VIN) == "" {
		return MetricSample{}, SourceMileageSample{}, false, nil
	}
	location, ok := telemetry.LocationProjectionForProtocol(env.Protocol, env.Fields)
	if !ok || location.SpeedKMH == nil || *location.SpeedKMH != 0 {
		return MetricSample{}, SourceMileageSample{}, false, nil
	}
	eventMS, _, ok := envelope.NormalizedEventTimeMSWithReason(env)
	if !ok {
		return MetricSample{}, SourceMileageSample{}, false, nil
	}
	eventTime := time.UnixMilli(eventMS).In(w.loc)
	sample := MetricSample{
		VIN:            strings.TrimSpace(env.VIN),
		Protocol:       env.Protocol,
		StatDate:       eventTime.Format("2006-01-02"),
		EventTime:      eventTime,
		SourceKey:      SourceKeyForSource(env.Protocol, env.Phone, env.DeviceID, identity.SourceIP, identity.SourceKind, identity.SourceCode),
		Phone:          strings.TrimSpace(env.Phone),
		DeviceID:       strings.TrimSpace(env.DeviceID),
		SourceEndpoint: strings.TrimSpace(env.SourceEndpoint),
		PlatformName:   strings.TrimSpace(env.PlatformName),
	}
	candidate := SourceMileageSampleFromMetric(sample, identity)
	baseline, found, err := w.previousBaseline(ctx, candidate)
	if err != nil || !found || baseline.LatestTotalKM <= 0 || baseline.LatestEventTime.IsZero() {
		return MetricSample{}, SourceMileageSample{}, false, err
	}
	sample.TotalMileageKM = baseline.LatestTotalKM
	candidate = SourceMileageSampleFromMetric(sample, identity)
	candidate.FirstTotalKM = baseline.LatestTotalKM
	candidate.LatestTotalKM = baseline.LatestTotalKM
	candidate.DailyKM = 0
	candidate.FirstEventTime = baseline.LatestEventTime
	candidate.LatestEventTime = eventTime
	candidate.QualityStatus = QualityOK
	candidate.QualityReason = QualityReasonStationaryCarry
	return sample, candidate, true, nil
}

func (w *Writer) writeMileageSample(ctx context.Context, sample MetricSample, candidate SourceMileageSample, projectDaily bool) error {
	if projectDaily {
		if beginner, ok := w.exec.(txBeginner); ok {
			tx, err := beginner.BeginTx(ctx, nil)
			if err != nil {
				return err
			}
			if err := UpsertSourceMileage(ctx, tx, candidate); err != nil {
				_ = tx.Rollback()
				return err
			}
			if ShouldNormalizePlatformSourceMileage(candidate) {
				if err := NormalizePlatformSourceMileage(ctx, tx, sample.VIN, sample.StatDate, sample.Protocol); err != nil {
					_ = tx.Rollback()
					return err
				}
			}
			if err := projectDailyMileageWithExec(ctx, tx, sample.VIN, sample.StatDate, sample.Protocol); err != nil {
				_ = tx.Rollback()
				return err
			}
			return tx.Commit()
		}
	}
	if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil {
		return err
	}
	if projectDaily {
		if ShouldNormalizePlatformSourceMileage(candidate) {
			if err := NormalizePlatformSourceMileage(ctx, w.exec, sample.VIN, sample.StatDate, sample.Protocol); err != nil {
				return err
			}
		}
		return ProjectDailyMileage(ctx, w.exec, sample.VIN, sample.StatDate, sample.Protocol)
	}
	return nil
}

func (w *Writer) sourceSeenAt(env envelope.FrameEnvelope) time.Time {
	if env.ReceivedAtMS > 0 {
		return time.UnixMilli(env.ReceivedAtMS).In(w.loc)
	}
	if eventMS, ok := statEventTimeMS(env); ok {
		return time.UnixMilli(eventMS).In(w.loc)
	}
	return time.Now().In(w.loc)
}

func (w *Writer) shouldTouchSource(identity SourceIdentity, seenAt time.Time) bool {
	key := string(identity.Protocol) + "|" + identity.SourceIP
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.sourceTouchInterval == 0 {
		return true
	}
	last, ok := w.lastSourceSeen[key]
	if ok && !seenAt.After(last.Add(w.sourceTouchInterval)) {
		return false
	}
	return true
}

func (w *Writer) markSourceTouched(identity SourceIdentity, seenAt time.Time) {
	key := string(identity.Protocol) + "|" + identity.SourceIP
	w.mu.Lock()
	w.lastSourceSeen[key] = seenAt
	w.enforceCacheLimitsLocked()
	w.mu.Unlock()
}

func (w *Writer) shouldProjectDailyMileage(sample MetricSample) bool {
	key := projectionCacheKey(sample)
	w.mu.Lock()
	defer w.mu.Unlock()
	interval := w.projectionInterval
	if interval == 0 {
		return true
	}
	entry, ok := w.lastProjection[key]
	if !ok {
		return true
	}
	if _, ok := entry.sourceKeys[sample.SourceKey]; !ok {
		return true
	}
	return sample.EventTime.After(entry.projectedAt.Add(interval))
}

func (w *Writer) markProjected(sample MetricSample) {
	key := projectionCacheKey(sample)
	prefix := projectionCachePrefix(sample)
	w.mu.Lock()
	defer w.mu.Unlock()
	w.deleteProjectionPrefixExceptLocked(prefix, key)
	entry := w.lastProjection[key]
	if entry.sourceKeys == nil {
		entry.sourceKeys = map[string]struct{}{}
	}
	entry.sourceKeys[sample.SourceKey] = struct{}{}
	if entry.projectedAt.IsZero() || sample.EventTime.After(entry.projectedAt) {
		entry.projectedAt = sample.EventTime
	}
	w.lastProjection[key] = entry
	if pending, ok := w.pendingProjection[key]; ok && !pending.sample.EventTime.After(sample.EventTime) {
		delete(w.pendingProjection, key)
	}
	w.addCacheKeyLocked(w.projectionKeysByPrefix, prefix, key)
	w.enforceCacheLimitsLocked()
}

func (w *Writer) markProjectionPending(sample MetricSample, queuedAt time.Time) {
	key := projectionCacheKey(sample)
	prefix := projectionCachePrefix(sample)
	w.mu.Lock()
	defer w.mu.Unlock()
	current, ok := w.pendingProjection[key]
	if !ok || sample.EventTime.After(current.sample.EventTime) ||
		(sample.EventTime.Equal(current.sample.EventTime) && queuedAt.After(current.queuedAt)) {
		w.pendingProjection[key] = pendingProjectionEntry{sample: sample, queuedAt: queuedAt}
	}
	w.addCacheKeyLocked(w.projectionKeysByPrefix, prefix, key)
	w.enforceCacheLimitsLocked()
}

// FlushPendingProjections closes the eventual-consistency gap introduced by
// projection throttling. Source candidates are persisted immediately, while
// final projections are flushed after their throttle window even when a
// vehicle stops reporting before another message can trigger projection.
func (w *Writer) FlushPendingProjections(ctx context.Context, readyBefore time.Time) (ProjectionFlushResult, error) {
	type pendingItem struct {
		key   string
		entry pendingProjectionEntry
	}

	w.mu.Lock()
	items := make([]pendingItem, 0, len(w.pendingProjection))
	for key, entry := range w.pendingProjection {
		if readyBefore.IsZero() || !entry.queuedAt.After(readyBefore) {
			items = append(items, pendingItem{key: key, entry: entry})
		}
	}
	w.mu.Unlock()
	sort.Slice(items, func(i, j int) bool {
		return items[i].key < items[j].key
	})

	var result ProjectionFlushResult
	var flushErr error
	for _, item := range items {
		if err := ctx.Err(); err != nil {
			return result, errors.Join(flushErr, err)
		}
		result.Attempted++
		if err := w.projectPendingMileage(ctx, item.entry.sample); err != nil {
			result.Failed++
			flushErr = errors.Join(flushErr, fmt.Errorf("project pending mileage %s: %w", item.key, err))
			continue
		}
		w.markProjected(item.entry.sample)
		result.Written++
	}
	return result, flushErr
}

// ReconcileDateProjections rebuilds final daily rows whose durable source
// candidates are newer than the projection (or whose projection is missing).
// The pending queue is intentionally in-memory for the hot path; this durable
// sweep closes the crash window between a committed source candidate and its
// throttled final projection.
func (w *Writer) ReconcileDateProjections(ctx context.Context, statDate string) (ProjectionReconcileResult, error) {
	var result ProjectionReconcileResult
	if w.query == nil {
		return result, errors.New("daily mileage projection reconciliation requires query support")
	}
	statDate = strings.TrimSpace(statDate)
	if statDate == "" {
		return result, errors.New("daily mileage projection reconciliation requires stat date")
	}
	rows, err := w.query.QueryContext(ctx, staleDailyMileageProjectionTargetsSQL, statDate)
	if err != nil {
		return result, err
	}
	type target struct {
		vin      string
		protocol envelope.Protocol
	}
	var targets []target
	for rows.Next() {
		var item target
		if err := rows.Scan(&item.vin, &item.protocol); err != nil {
			_ = rows.Close()
			return result, err
		}
		if strings.TrimSpace(item.vin) != "" && strings.TrimSpace(string(item.protocol)) != "" {
			targets = append(targets, item)
		}
	}
	if err := rows.Close(); err != nil {
		return result, err
	}
	if err := rows.Err(); err != nil {
		return result, err
	}
	result.Targets = len(targets)
	var reconcileErr error
	for _, item := range targets {
		if err := ctx.Err(); err != nil {
			return result, errors.Join(reconcileErr, err)
		}
		result.Attempted++
		if err := ProjectDailyMileage(ctx, w.exec, item.vin, statDate, item.protocol); err != nil {
			result.Failed++
			reconcileErr = errors.Join(reconcileErr, fmt.Errorf("reconcile daily mileage %s/%s/%s: %w", item.vin, statDate, item.protocol, err))
			continue
		}
		result.Written++
	}
	return result, reconcileErr
}

func (w *Writer) projectPendingMileage(ctx context.Context, sample MetricSample) error {
	project := func(exec Execer) error {
		if strings.Contains(sample.SourceKey, platformSourceKeyPrefix) {
			if err := NormalizePlatformSourceMileage(ctx, exec, sample.VIN, sample.StatDate, sample.Protocol); err != nil {
				return err
			}
		}
		return projectDailyMileageWithExec(ctx, exec, sample.VIN, sample.StatDate, sample.Protocol)
	}
	if beginner, ok := w.exec.(txBeginner); ok {
		tx, err := beginner.BeginTx(ctx, nil)
		if err != nil {
			return err
		}
		if err := project(tx); err != nil {
			_ = tx.Rollback()
			return err
		}
		return tx.Commit()
	}
	return project(w.exec)
}

const staleDailyMileageProjectionTargetsSQL = `
SELECT s.vin, s.protocol
FROM vehicle_daily_mileage_source s
LEFT JOIN vehicle_daily_mileage m
  ON m.vin = s.vin
 AND m.stat_date = s.stat_date
 AND m.protocol = s.protocol
WHERE s.stat_date = ?
  AND s.quality_status = '` + QualityOK + `'
  AND (
    m.vin IS NULL
    OR s.updated_at > m.updated_at
    OR s.latest_event_time > m.updated_at
  )
GROUP BY s.vin, s.protocol
ORDER BY s.protocol ASC, s.vin ASC
`

func (w *Writer) seenSameMileage(sample MetricSample) bool {
	key := mileageCacheKey(sample)
	w.mu.Lock()
	defer w.mu.Unlock()
	if last, ok := w.lastTotalMileage[key]; ok && last == sample.TotalMileageKM {
		return true
	}
	return false
}

func (w *Writer) markMileageWritten(sample MetricSample) {
	prefix := mileageCachePrefix(sample)
	key := mileageCacheKey(sample)
	w.mu.Lock()
	defer w.mu.Unlock()
	w.deleteMileagePrefixExceptLocked(prefix, key)
	w.deleteBaselinePrefixExceptLocked(prefix, key)
	w.lastTotalMileage[key] = sample.TotalMileageKM
	w.addCacheKeyLocked(w.mileageKeysByPrefix, prefix, key)
	w.enforceCacheLimitsLocked()
}

func (w *Writer) maybeCleanupCaches(now time.Time) {
	if now.IsZero() {
		now = time.Now().In(w.loc)
	}
	w.mu.Lock()
	defer w.mu.Unlock()
	retention := w.cacheRetention
	if retention <= 0 {
		w.enforceCacheLimitsLocked()
		return
	}
	interval := w.cacheCleanupInterval
	if interval > 0 && !w.lastCacheCleanup.IsZero() && !now.After(w.lastCacheCleanup.Add(interval)) {
		w.enforceCacheLimitsLocked()
		return
	}
	w.lastCacheCleanup = now
	cleanupStats := cacheCleanupStats{}
	cutoff := now.Add(-retention)
	cutoffDate := cutoff.In(w.loc).Format("2006-01-02")
	for key, seenAt := range w.lastSourceSeen {
		if !seenAt.IsZero() && seenAt.Before(cutoff) {
			delete(w.lastSourceSeen, key)
			cleanupStats.sourceSeen++
		}
	}
	for key, entry := range w.lastProjection {
		if entry.projectedAt.IsZero() || entry.projectedAt.Before(cutoff) || cacheKeyDateBefore(key, cutoffDate) {
			w.deleteProjectionKeyLocked(key)
			cleanupStats.projection++
		}
	}
	for key := range w.lastTotalMileage {
		if cacheKeyDateBefore(key, cutoffDate) {
			w.deleteMileageKeyLocked(key)
			cleanupStats.totalMileage++
		}
	}
	for key := range w.baselineCache {
		if cacheKeyDateBefore(key, cutoffDate) {
			w.deleteBaselineKeyLocked(key)
			cleanupStats.baseline++
		}
	}
	for key, eventTime := range w.lastGPSAccumulation {
		if eventTime.IsZero() || eventTime.Before(cutoff) {
			delete(w.lastGPSAccumulation, key)
			cleanupStats.gps++
		}
	}
	w.lastCacheCleanupStats = cleanupStats
	w.enforceCacheLimitsLocked()
}

func (w *Writer) applyRealtimeBaseline(ctx context.Context, candidate *SourceMileageSample) error {
	if candidate == nil {
		return nil
	}
	baseline, found, err := w.previousBaseline(ctx, *candidate)
	if err != nil {
		return err
	}
	if !found {
		// If no earlier odometer exists at all, use the first current-day sample.
		// Later samples retain that boundary through the in-memory baseline cache.
		candidate.DailyKM = 0
		candidate.QualityStatus = QualityOK
		candidate.QualityReason = QualityReasonCurrentDayFirst
		return nil
	}
	candidate.FirstTotalKM = baseline.LatestTotalKM
	candidate.FirstEventTime = baseline.LatestEventTime
	candidate.DailyKM = DailyMileageFromDayBoundary(baseline.LatestTotalKM, candidate.LatestTotalKM)
	candidate.QualityStatus = QualityOK
	candidate.QualityReason = baseline.QualityReason
	if candidate.QualityReason == "" {
		candidate.QualityReason = QualityReasonHistorical
	}
	ApplyMileageQualityRules(candidate)
	return nil
}

func (w *Writer) previousBaseline(ctx context.Context, candidate SourceMileageSample) (sourceBaseline, bool, error) {
	cacheKey := mileageCacheKey(MetricSample{
		VIN:       candidate.VIN,
		Protocol:  candidate.Protocol,
		StatDate:  candidate.StatDate,
		SourceKey: candidate.SourceKey,
	})
	now := time.Now()
	w.mu.Lock()
	if cached, ok := w.baselineCache[cacheKey]; ok {
		ttl := w.baselineMissTTL
		if cached.found {
			ttl = w.baselineHitTTL
		}
		expired := ttl == 0 || cached.cachedAt.IsZero() || now.Sub(cached.cachedAt) >= ttl
		if !expired {
			w.mu.Unlock()
			return cached.baseline, cached.found, nil
		}
		w.deleteBaselineKeyLocked(cacheKey)
	}
	w.mu.Unlock()

	baseline, found, err := lookupPreviousSourceBaseline(ctx, w.query, candidate.VIN, candidate.StatDate, candidate.Protocol, candidate.SourceKey)
	if err != nil {
		return sourceBaseline{}, false, err
	}

	if found {
		w.cacheBaseline(candidate, cacheKey, sourceBaselineCacheEntry{baseline: baseline, found: true})
	} else {
		w.cacheBaseline(candidate, cacheKey, sourceBaselineCacheEntry{found: false})
	}
	return baseline, found, nil
}

func (w *Writer) markBaselineWritten(sample MetricSample, candidate SourceMileageSample) {
	if candidate.QualityStatus != QualityOK {
		return
	}
	baselineTotal := candidate.FirstTotalKM
	if baselineTotal <= 0 {
		baselineTotal = candidate.LatestTotalKM
	}
	baselineTime := candidate.FirstEventTime
	if baselineTime.IsZero() {
		baselineTime = candidate.LatestEventTime
	}
	if baselineTotal <= 0 || baselineTime.IsZero() {
		return
	}
	cacheKey := mileageCacheKey(sample)
	w.cacheBaseline(candidate, cacheKey, sourceBaselineCacheEntry{
		baseline: sourceBaseline{
			LatestTotalKM:   baselineTotal,
			LatestEventTime: baselineTime,
			QualityReason:   candidate.QualityReason,
		},
		found: true,
	})
}

func (w *Writer) cacheBaseline(candidate SourceMileageSample, cacheKey string, entry sourceBaselineCacheEntry) {
	if cacheKey == "" {
		return
	}
	w.mu.Lock()
	if previous, ok := w.baselineCache[cacheKey]; ok &&
		previous.found && entry.found &&
		previous.baseline.LatestTotalKM == entry.baseline.LatestTotalKM &&
		previous.baseline.LatestEventTime.Equal(entry.baseline.LatestEventTime) &&
		!previous.cachedAt.IsZero() {
		// markBaselineWritten runs for every accepted sample. Preserve the
		// original cache age when the durable day-boundary baseline has not
		// changed, otherwise an active vehicle would keep stale history forever.
		entry.cachedAt = previous.cachedAt
	}
	if entry.cachedAt.IsZero() {
		entry.cachedAt = time.Now()
	}
	w.baselineCache[cacheKey] = entry
	w.addCacheKeyLocked(w.baselineKeysByPrefix, mileageCachePrefix(MetricSample{
		VIN:       candidate.VIN,
		Protocol:  candidate.Protocol,
		SourceKey: candidate.SourceKey,
	}), cacheKey)
	w.enforceCacheLimitsLocked()
	w.mu.Unlock()
}

func (w *Writer) addCacheKeyLocked(index map[string]map[string]struct{}, prefix string, key string) {
	if prefix == "" || key == "" {
		return
	}
	keys := index[prefix]
	if keys == nil {
		keys = map[string]struct{}{}
		index[prefix] = keys
	}
	keys[key] = struct{}{}
}

func (w *Writer) deleteMileagePrefixExceptLocked(prefix string, keep string) {
	for key := range w.mileageKeysByPrefix[prefix] {
		if key == keep {
			continue
		}
		w.deleteMileageKeyLocked(key)
	}
}

func (w *Writer) deleteBaselinePrefixExceptLocked(prefix string, keep string) {
	for key := range w.baselineKeysByPrefix[prefix] {
		if key == keep {
			continue
		}
		w.deleteBaselineKeyLocked(key)
	}
}

func (w *Writer) deleteProjectionPrefixExceptLocked(prefix string, keep string) {
	for key := range w.projectionKeysByPrefix[prefix] {
		if key == keep {
			continue
		}
		w.deleteProjectionKeyLocked(key)
	}
}

func (w *Writer) deleteMileageKeyLocked(key string) {
	delete(w.lastTotalMileage, key)
	w.deleteIndexedCacheKeyLocked(w.mileageKeysByPrefix, cacheKeyPrefix(key), key)
}

func (w *Writer) deleteBaselineKeyLocked(key string) {
	delete(w.baselineCache, key)
	w.deleteIndexedCacheKeyLocked(w.baselineKeysByPrefix, cacheKeyPrefix(key), key)
}

func (w *Writer) deleteProjectionKeyLocked(key string) {
	delete(w.lastProjection, key)
	delete(w.pendingProjection, key)
	w.deleteIndexedCacheKeyLocked(w.projectionKeysByPrefix, cacheKeyPrefix(key), key)
}

func (w *Writer) deleteIndexedCacheKeyLocked(index map[string]map[string]struct{}, prefix string, key string) {
	if prefix == "" || key == "" {
		return
	}
	keys := index[prefix]
	if len(keys) == 0 {
		return
	}
	delete(keys, key)
	if len(keys) == 0 {
		delete(index, prefix)
	}
}

func (w *Writer) enforceCacheLimitsLocked() {
	if w.maxCacheEntries <= 0 {
		return
	}
	for len(w.lastTotalMileage) > w.maxCacheEntries {
		victim := oldestCacheKeyByDate(w.lastTotalMileage)
		if victim == "" {
			break
		}
		w.deleteMileageKeyLocked(victim)
		w.cacheEvictions.totalMileage++
	}
	for len(w.baselineCache) > w.maxCacheEntries {
		victim := oldestCacheKeyByDate(w.baselineCache)
		if victim == "" {
			break
		}
		w.deleteBaselineKeyLocked(victim)
		w.cacheEvictions.baseline++
	}
	for len(w.lastProjection) > w.maxCacheEntries {
		victim := oldestProjectionCacheKey(w.lastProjection)
		if victim == "" {
			break
		}
		w.deleteProjectionKeyLocked(victim)
		w.cacheEvictions.projection++
	}
	for len(w.lastSourceSeen) > w.maxCacheEntries {
		victim := oldestSourceSeenKey(w.lastSourceSeen)
		if victim == "" {
			break
		}
		delete(w.lastSourceSeen, victim)
		w.cacheEvictions.sourceSeen++
	}
	for len(w.lastGPSAccumulation) > w.maxCacheEntries {
		victim := oldestSourceSeenKey(w.lastGPSAccumulation)
		if victim == "" {
			break
		}
		delete(w.lastGPSAccumulation, victim)
		w.cacheEvictions.gps++
	}
}

func mileageCachePrefix(sample MetricSample) string {
	return fmt.Sprintf("%s|%s|%s|", sample.VIN, sample.Protocol, sample.SourceKey)
}

func mileageCacheKey(sample MetricSample) string {
	return mileageCachePrefix(sample) + sample.StatDate
}

func projectionCacheKey(sample MetricSample) string {
	return projectionCachePrefix(sample) + sample.StatDate
}

func projectionCachePrefix(sample MetricSample) string {
	return fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol)
}

func cacheKeyDateBefore(key string, cutoffDate string) bool {
	date := cacheKeyDate(key)
	if len(date) != len("2006-01-02") {
		return false
	}
	return date < cutoffDate
}

func cacheKeyPrefix(key string) string {
	index := strings.LastIndex(key, "|")
	if index < 0 {
		return ""
	}
	return key[:index+1]
}

func cacheKeyDate(key string) string {
	index := strings.LastIndex(key, "|")
	if index < 0 || index == len(key)-1 {
		return ""
	}
	return key[index+1:]
}

func oldestCacheKeyByDate[T any](items map[string]T) string {
	victim := ""
	victimDate := ""
	for key := range items {
		date := cacheKeyDate(key)
		if date == "" {
			date = "9999-99-99"
		}
		if victim == "" || date < victimDate || (date == victimDate && key < victim) {
			victim = key
			victimDate = date
		}
	}
	return victim
}

func oldestProjectionCacheKey(items map[string]projectionCacheEntry) string {
	victim := ""
	var victimTime time.Time
	for key, entry := range items {
		if victim == "" ||
			(!entry.projectedAt.IsZero() && (victimTime.IsZero() || entry.projectedAt.Before(victimTime))) ||
			(entry.projectedAt.Equal(victimTime) && key < victim) {
			victim = key
			victimTime = entry.projectedAt
		}
	}
	if victim != "" {
		return victim
	}
	return oldestCacheKeyByDate(items)
}

func oldestSourceSeenKey(items map[string]time.Time) string {
	victim := ""
	var victimTime time.Time
	for key, seenAt := range items {
		if victim == "" ||
			(!seenAt.IsZero() && (victimTime.IsZero() || seenAt.Before(victimTime))) ||
			(seenAt.Equal(victimTime) && key < victim) {
			victim = key
			victimTime = seenAt
		}
	}
	return victim
}

type projectionCacheEntry struct {
	projectedAt time.Time
	sourceKeys  map[string]struct{}
}

type pendingProjectionEntry struct {
	sample   MetricSample
	queuedAt time.Time
}

type sourceBaselineCacheEntry struct {
	baseline sourceBaseline
	found    bool
	cachedAt time.Time
}

func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) {
	samples, _, err := samplesFromEnvelopeWithResult(env, loc)
	return samples, err
}

func samplesFromEnvelopeWithResult(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, AppendResult, error) {
	var result AppendResult
	if len(env.Fields) == 0 {
		result.SamplesSkippedMissingFields = 1
		return nil, result, nil
	}
	vin := strings.TrimSpace(env.VIN)
	if vin == "" {
		result.SamplesSkippedMissingVIN = 1
		return nil, result, nil
	}
	totalMileage, ok := totalMileageKMFromEnvelope(env)
	if !ok {
		if isMileageCandidateEnvelope(env) {
			result.SamplesSkippedMissingMileage = 1
		} else {
			result.SamplesSkippedNonMileageFrame = 1
		}
		return nil, result, nil
	}
	if totalMileage <= 0 {
		result.SamplesSkippedNonPositiveMileage = 1
		return nil, result, nil
	}
	if loc == nil {
		loc = time.FixedZone("Asia/Shanghai", 8*3600)
	}
	eventMS, reason, ok := envelope.NormalizedEventTimeMSWithReason(env)
	if !ok {
		result.SamplesSkippedMissingTime = 1
		return nil, result, nil
	}
	if reason == envelope.EventTimeReasonReceivedFutureEvent {
		result.SamplesAdjustedFutureEventTime = 1
	}
	statDate := time.UnixMilli(eventMS).In(loc).Format("2006-01-02")
	result.SamplesFound = 1
	return []MetricSample{{
		VIN:            vin,
		Protocol:       env.Protocol,
		StatDate:       statDate,
		TotalMileageKM: totalMileage,
		EventTime:      time.UnixMilli(eventMS).In(loc),
		SourceKey:      sourceKey(env),
		Phone:          strings.TrimSpace(env.Phone),
		DeviceID:       strings.TrimSpace(env.DeviceID),
		SourceEndpoint: strings.TrimSpace(env.SourceEndpoint),
		PlatformName:   strings.TrimSpace(env.PlatformName),
	}}, result, nil
}

func statEventTimeMS(env envelope.FrameEnvelope) (int64, bool) {
	return envelope.NormalizedEventTimeMS(env)
}

func sourceKey(env envelope.FrameEnvelope) string {
	return SourceKeyForSource(env.Protocol, env.Phone, env.DeviceID, NormalizeSourceIP(env.SourceEndpoint), env.SourceKind, env.SourceCode)
}

func isIgnorableSchemaChangeError(err error) bool {
	if err == nil {
		return false
	}
	text := strings.ToLower(err.Error())
	return strings.Contains(text, "duplicate column") ||
		strings.Contains(text, "1060") ||
		strings.Contains(text, "duplicate key name") ||
		strings.Contains(text, "1061") ||
		strings.Contains(text, "can't drop") ||
		strings.Contains(text, "1091")
}

type mileageFieldMapping = telemetry.MileageFieldMapping

func totalMileageKMFromEnvelope(env envelope.FrameEnvelope) (float64, bool) {
	return telemetry.TotalMileageKM(env.Protocol, env.Fields)
}

func mileageMappingsByProtocol(protocol envelope.Protocol) []mileageFieldMapping {
	return telemetry.MileageFieldMappings(protocol)
}

func isMileageCandidateEnvelope(env envelope.FrameEnvelope) bool {
	switch env.Protocol {
	case envelope.ProtocolJT808:
		messageID := strings.TrimSpace(env.MessageID)
		return strings.EqualFold(messageID, "0x0200") ||
			hasFieldPrefix(env.Fields, "jt808.location.") ||
			hasAnyField(env.Fields, envelope.FieldLatitude, envelope.FieldLongitude, envelope.FieldSpeedKMH)
	case envelope.ProtocolGB32960:
		return hasFieldPrefix(env.Fields, "gb32960.vehicle.") ||
			hasAnyField(env.Fields, envelope.FieldLatitude, envelope.FieldLongitude, envelope.FieldSpeedKMH, envelope.FieldSOCPercent)
	case envelope.ProtocolYutongMQTT:
		return hasFieldPrefix(env.Fields, "yutong_mqtt.data.") ||
			hasFieldPrefix(env.Fields, "yutong_mqtt.root.data.") ||
			hasAnyField(env.Fields, envelope.FieldLatitude, envelope.FieldLongitude, envelope.FieldSpeedKMH, envelope.FieldSOCPercent)
	default:
		return true
	}
}

func hasFieldPrefix(fields map[string]any, prefix string) bool {
	if len(fields) == 0 || prefix == "" {
		return false
	}
	for key := range fields {
		if strings.HasPrefix(key, prefix) {
			return true
		}
	}
	return false
}

func hasAnyField(fields map[string]any, keys ...string) bool {
	if len(fields) == 0 {
		return false
	}
	for _, key := range keys {
		if _, ok := fields[key]; ok {
			return true
		}
	}
	return false
}

func floatField(env envelope.FrameEnvelope, key string) (float64, bool) {
	if env.Fields == nil {
		return 0, false
	}
	value, ok := env.Fields[key]
	if !ok || value == nil {
		return 0, false
	}
	switch typed := value.(type) {
	case float64:
		return typed, true
	case float32:
		return float64(typed), true
	case int:
		return float64(typed), true
	case int64:
		return float64(typed), true
	case uint16:
		return float64(typed), true
	case uint32:
		return float64(typed), true
	case string:
		parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64)
		return parsed, err == nil
	default:
		return 0, false
	}
}