lingniu-vehicle-ingest/vehicle-data-platform/apps/api/internal/openplatform/mileage_reconciliation.go

package openplatform

import (
	"context"
	"fmt"
	"math"
	"sort"
	"strings"
	"time"
)

// ReconciledMileageRange uses terminal odometers, including one preceding
// observation per protocol. The same function serves single days and every
// range page: neither the requested start date nor page size changes a result.
func (r *MySQLRepository) ReconciledMileageRange(ctx context.Context, vins []string, start, end string, protocols []string) (map[string]DailyMileage, error) {
	if len(vins) == 0 {
		return map[string]DailyMileage{}, nil
	}
	if len(protocols) == 0 {
		protocols = []string{"GB32960", "YUTONG_MQTT", "JT808"}
	}
	placeholders := func(n int) string { return strings.TrimRight(strings.Repeat("?,", n), ",") }
	// Unknown-protocol legacy imports and GPS distance are not terminal odometers.
	eligible := func(alias string) string {
		return fmt.Sprintf(`%[1]s.quality_status IN ('OK','INVALID_DELTA')
 AND %[1]s.latest_total_mileage_km > 0
 AND COALESCE(%[1]s.quality_reason,'') <> 'gps_coordinate_accumulation'
 AND %[1]s.source_ip NOT IN ('legacy-mysql.lingniu-prod','manual-lingniu-prod-day-mileage')
 AND %[1]s.latest_event_time IS NOT NULL
 AND %[1]s.latest_event_time < TIMESTAMP(%[1]s.stat_date)+INTERVAL 1 DAY`, alias)
	}
	filter := `vin IN (` + placeholders(len(vins)) + `) AND protocol IN (` + placeholders(len(protocols)) + `)`
	query := `WITH wanted AS (
 SELECT DISTINCT vin,protocol,stat_date FROM vehicle_daily_mileage_source p
 WHERE ` + filter + ` AND stat_date BETWEEN ? AND ? AND ` + eligible("p") + `
 UNION ALL
 SELECT vin,protocol,MAX(stat_date) AS stat_date FROM vehicle_daily_mileage_source p
 WHERE ` + filter + ` AND stat_date<? AND ` + eligible("p") + ` GROUP BY vin,protocol
 )
 SELECT s.vin,DATE_FORMAT(s.stat_date,'%Y-%m-%d'),s.protocol,s.source_key,
 s.daily_mileage_km,s.latest_total_mileage_km,
 DATE_FORMAT(s.latest_event_time,'%Y-%m-%dT%H:%i:%s+08:00'),
 DATE_FORMAT(s.updated_at,'%Y-%m-%dT%H:%i:%s+08:00'),
 COALESCE(DATE_FORMAT(s.first_event_time,'%Y-%m-%dT%H:%i:%s+08:00'),''),
 CASE WHEN s.quality_status='INVALID_DELTA' THEN COALESCE(s.quality_reason,'INVALID_DELTA') ELSE '' END
 FROM wanted w JOIN vehicle_daily_mileage_source s
 ON s.vin=w.vin AND s.protocol=w.protocol AND s.stat_date=w.stat_date
 WHERE ` + eligible("s") + `
 ORDER BY s.stat_date,s.vin,s.protocol,s.is_selected DESC,
 CASE WHEN s.quality_status='OK' THEN 0 ELSE 1 END,s.latest_event_time DESC,s.sample_count DESC,s.source_key`
	args := make([]any, 0, 2*(len(vins)+len(protocols))+3)
	for _, v := range vins {
		args = append(args, v)
	}
	for _, p := range protocols {
		args = append(args, p)
	}
	args = append(args, start, end)
	for _, v := range vins {
		args = append(args, v)
	}
	for _, p := range protocols {
		args = append(args, p)
	}
	args = append(args, start)
	rows, err := r.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var points []DailyMileage
	seen := map[string]bool{}
	for rows.Next() {
		var v DailyMileage
		if err := rows.Scan(&v.VIN, &v.Date, &v.Protocol, &v.SourceKey, &v.MileageKm, &v.TotalMileageKm, &v.DataTime, &v.UpdatedAt, &v.StatisticsStartTime, &v.DataQuality); err != nil {
			return nil, err
		}
		key := v.VIN + "|" + v.Date + "|" + v.Protocol
		if !seen[key] {
			points = append(points, v)
			seen[key] = true
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return reconcileMileage(points, vins, start, end, protocols)
}

func reconcileMileage(points []DailyMileage, vins []string, start, end string, protocols []string) (map[string]DailyMileage, error) {
	loc := time.FixedZone("Asia/Shanghai", 8*3600)
	first, err := time.ParseInLocation("2006-01-02", start, loc)
	if err != nil {
		return nil, err
	}
	last, err := time.ParseInLocation("2006-01-02", end, loc)
	if err != nil {
		return nil, err
	}
	if last.Before(first) {
		return nil, fmt.Errorf("invalid mileage interval")
	}
	if len(protocols) == 0 {
		protocols = []string{"GB32960", "YUTONG_MQTT", "JT808"}
	}
	sort.SliceStable(points, func(i, j int) bool { return points[i].Date < points[j].Date })
	state := map[string]map[string]DailyMileage{}
	pick := func(vin string) (DailyMileage, bool) {
		for _, p := range protocols {
			if v, ok := state[vin][p]; ok {
				return v, true
			}
		}
		return DailyMileage{}, false
	}
	put := func(v DailyMileage) {
		if state[v.VIN] == nil {
			state[v.VIN] = map[string]DailyMileage{}
		}
		state[v.VIN][v.Protocol] = v
	}
	i := 0
	for i < len(points) && points[i].Date < start {
		put(points[i])
		i++
	}
	out := map[string]DailyMileage{}
	for day := first; !day.After(last); day = day.AddDate(0, 0, 1) {
		date := day.Format("2006-01-02")
		before := map[string]DailyMileage{}
		for _, vin := range vins {
			if p, ok := pick(vin); ok {
				before[vin] = p
			}
		}
		for i < len(points) && points[i].Date == date {
			put(points[i])
			i++
		}
		for _, vin := range vins {
			current, ok := pick(vin)
			if !ok {
				continue
			}
			result := current
			previous, hasPrevious := before[vin]
			if current.Date < date {
				result.MileageKm = 0
				result.StatisticsStartTime = ""
				if result.DataQuality == "" {
					result.DataQuality = "CARRIED_FORWARD"
				}
			} else if current.DataQuality != "" {
				result.MileageKm = 0
			} else if !hasPrevious {
				result.DataQuality = "NO_PREVIOUS_BASELINE"
			} else if current.Protocol != previous.Protocol || current.SourceKey != previous.SourceKey {
				result.DataQuality = "ODOMETER_SOURCE_CHANGED"
			} else if current.TotalMileageKm < previous.TotalMileageKm {
				result.DataQuality = mileageTotalRollbackQuality
			} else if previous.DataQuality != "" {
				result.DataQuality = "PREVIOUS_ODOMETER_ANOMALY"
			} else {
				result.MileageKm = math.Round((current.TotalMileageKm-previous.TotalMileageKm)*1000) / 1000
				result.StatisticsStartTime = previous.DataTime
			}
			out[dailyMileageKey(vin, date)] = result
		}
	}
	return out, nil
}