773 lines
29 KiB
Go
773 lines
29 KiB
Go
package stats
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"strings"
|
|
"time"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
)
|
|
|
|
const (
|
|
QualityOK = "OK"
|
|
QualityNoPreviousBaseline = "NO_PREVIOUS_BASELINE"
|
|
QualityInvalidDelta = "INVALID_DELTA"
|
|
QualityReasonHistorical = "historical_source_baseline"
|
|
QualityReasonCurrentDayFirst = "current_day_first_baseline"
|
|
maxSelectedDailyMileageKM = 2500
|
|
maxSelectedDailyMileageKMSQL = "2500"
|
|
maxNegativeMileageJitterKMSQL = "1"
|
|
directSourceKeySuffix = "@DIRECT"
|
|
platformSourceKeyPrefix = "@PLATFORM:"
|
|
)
|
|
|
|
type SourceMileageSample struct {
|
|
VIN string
|
|
StatDate string
|
|
Protocol envelope.Protocol
|
|
SourceKey string
|
|
SourceIP string
|
|
SourceEndpoint string
|
|
Phone string
|
|
DeviceID string
|
|
PlatformName string
|
|
FirstTotalKM float64
|
|
LatestTotalKM float64
|
|
DailyKM float64
|
|
SampleCount int64
|
|
FirstEventTime time.Time
|
|
LatestEventTime time.Time
|
|
QualityStatus string
|
|
QualityReason string
|
|
}
|
|
|
|
func SourceKey(protocol envelope.Protocol, phone string, deviceID string, sourceIP string) string {
|
|
identity, _ := sourceKeyIdentity(phone, deviceID)
|
|
return string(protocol) + ":" + identity + "@" + strings.TrimSpace(sourceIP)
|
|
}
|
|
|
|
func sourceKeyIdentity(phone string, deviceID string) (string, bool) {
|
|
identity := strings.TrimSpace(phone)
|
|
if identity == "" {
|
|
identity = strings.TrimSpace(deviceID)
|
|
}
|
|
if identity == "" {
|
|
return "unknown", false
|
|
}
|
|
return identity, true
|
|
}
|
|
|
|
func SourceKeyForKind(protocol envelope.Protocol, phone string, deviceID string, sourceIP string, sourceKind string) string {
|
|
return SourceKeyForSource(protocol, phone, deviceID, sourceIP, sourceKind, "")
|
|
}
|
|
|
|
func SourceKeyForSource(protocol envelope.Protocol, phone string, deviceID string, sourceIP string, sourceKind string, sourceCode string) string {
|
|
identity, hasIdentity := sourceKeyIdentity(phone, deviceID)
|
|
if strings.EqualFold(strings.TrimSpace(sourceKind), "DIRECT") && hasIdentity {
|
|
return string(protocol) + ":" + identity + directSourceKeySuffix
|
|
}
|
|
if strings.EqualFold(strings.TrimSpace(sourceKind), "PLATFORM") && hasIdentity {
|
|
sourceCode = strings.TrimSpace(sourceCode)
|
|
if sourceCode != "" {
|
|
return string(protocol) + ":" + identity + platformSourceKeyPrefix + sourceCode
|
|
}
|
|
}
|
|
return SourceKey(protocol, phone, deviceID, sourceIP)
|
|
}
|
|
|
|
func SourceMileageSampleFromMetric(sample MetricSample, identity SourceIdentity) SourceMileageSample {
|
|
eventTime := sample.EventTime
|
|
return SourceMileageSample{
|
|
VIN: sample.VIN,
|
|
StatDate: sample.StatDate,
|
|
Protocol: sample.Protocol,
|
|
SourceKey: SourceKeyForSource(sample.Protocol, sample.Phone, sample.DeviceID, identity.SourceIP, identity.SourceKind, identity.SourceCode),
|
|
SourceIP: identity.SourceIP,
|
|
SourceEndpoint: identity.SourceEndpoint,
|
|
Phone: sample.Phone,
|
|
DeviceID: sample.DeviceID,
|
|
PlatformName: firstNonEmpty(sample.PlatformName, identity.PlatformName),
|
|
FirstTotalKM: sample.TotalMileageKM,
|
|
LatestTotalKM: sample.TotalMileageKM,
|
|
DailyKM: 0,
|
|
SampleCount: 1,
|
|
FirstEventTime: eventTime,
|
|
LatestEventTime: eventTime,
|
|
QualityStatus: QualityOK,
|
|
QualityReason: "realtime_sample",
|
|
}
|
|
}
|
|
|
|
func firstNonEmpty(values ...string) string {
|
|
for _, value := range values {
|
|
value = strings.TrimSpace(value)
|
|
if value != "" {
|
|
return value
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func NormalizeDailyMileageDelta(deltaKM float64) (float64, bool, string) {
|
|
return NormalizeDailyMileageDeltaForWindow(deltaKM, time.Time{}, time.Time{})
|
|
}
|
|
|
|
// DailyMileageFromDayBoundary keeps the business formula explicit: the
|
|
// current day's latest cumulative odometer minus the nearest earlier
|
|
// cumulative odometer from the same source.
|
|
func DailyMileageFromDayBoundary(previousBaselineKM float64, currentDayLatestKM float64) float64 {
|
|
return currentDayLatestKM - previousBaselineKM
|
|
}
|
|
|
|
func NormalizeDailyMileageDeltaForWindow(deltaKM float64, firstEventTime time.Time, latestEventTime time.Time) (float64, bool, string) {
|
|
if deltaKM < 0 && deltaKM >= -maxNegativeMileageJitterKM {
|
|
return 0, true, "negative_jitter_clamped"
|
|
}
|
|
maxMileage := float64(MileageQualityLimitKM(firstEventTime, latestEventTime))
|
|
if deltaKM < 0 || deltaKM > maxMileage {
|
|
return deltaKM, false, "outside_daily_range"
|
|
}
|
|
return deltaKM, true, ""
|
|
}
|
|
|
|
func MileageQualityWindowDays(firstEventTime time.Time, latestEventTime time.Time) int {
|
|
if firstEventTime.IsZero() || latestEventTime.IsZero() || !latestEventTime.After(firstEventTime) {
|
|
return 1
|
|
}
|
|
location := latestEventTime.Location()
|
|
firstYear, firstMonth, firstDay := firstEventTime.In(location).Date()
|
|
latestYear, latestMonth, latestDay := latestEventTime.Date()
|
|
firstDate := time.Date(firstYear, firstMonth, firstDay, 0, 0, 0, 0, time.UTC)
|
|
latestDate := time.Date(latestYear, latestMonth, latestDay, 0, 0, 0, 0, time.UTC)
|
|
days := int(latestDate.Sub(firstDate) / (24 * time.Hour))
|
|
if days < 1 {
|
|
return 1
|
|
}
|
|
return days
|
|
}
|
|
|
|
func MileageQualityLimitKM(firstEventTime time.Time, latestEventTime time.Time) int {
|
|
// When the previous natural day is missing, the business baseline walks
|
|
// farther back. Scale the plausibility guard by that calendar-day gap while
|
|
// preserving the exact cumulative-odometer difference in the current row.
|
|
return maxSelectedDailyMileageKM * MileageQualityWindowDays(firstEventTime, latestEventTime)
|
|
}
|
|
|
|
func ApplyMileageQualityRules(sample *SourceMileageSample) {
|
|
if sample == nil {
|
|
return
|
|
}
|
|
if sample.QualityStatus == "" {
|
|
sample.QualityStatus = QualityOK
|
|
}
|
|
if sample.QualityStatus != QualityOK {
|
|
return
|
|
}
|
|
normalized, ok, reason := NormalizeDailyMileageDeltaForWindow(sample.DailyKM, sample.FirstEventTime, sample.LatestEventTime)
|
|
sample.DailyKM = normalized
|
|
if reason != "" {
|
|
sample.QualityReason = reason
|
|
}
|
|
if !ok {
|
|
sample.QualityStatus = QualityInvalidDelta
|
|
}
|
|
}
|
|
|
|
func UpsertSourceMileage(ctx context.Context, exec Execer, sample SourceMileageSample) error {
|
|
if exec == nil {
|
|
panic("stats execer must not be nil")
|
|
}
|
|
if sample.VIN == "" || sample.SourceKey == "" || strings.TrimSpace(sample.SourceIP) == "" {
|
|
return nil
|
|
}
|
|
if sample.QualityStatus == "" {
|
|
sample.QualityStatus = QualityOK
|
|
}
|
|
// MySQL DATETIME(0) may round 23:59:59.xxx into the next natural day.
|
|
// Truncate protocol event timestamps before persistence so day boundaries
|
|
// remain stable and match the TDengine event_time predicate.
|
|
sample.FirstEventTime = truncateMileageEventTime(sample.FirstEventTime)
|
|
sample.LatestEventTime = truncateMileageEventTime(sample.LatestEventTime)
|
|
_, err := exec.ExecContext(ctx, upsertSourceMileageSQL,
|
|
sample.VIN,
|
|
sample.StatDate,
|
|
string(sample.Protocol),
|
|
sample.SourceKey,
|
|
sample.SourceIP,
|
|
sample.SourceEndpoint,
|
|
sample.Phone,
|
|
sample.DeviceID,
|
|
sample.PlatformName,
|
|
sample.FirstTotalKM,
|
|
sample.LatestTotalKM,
|
|
sample.DailyKM,
|
|
sample.SampleCount,
|
|
sample.FirstEventTime,
|
|
sample.LatestEventTime,
|
|
sample.QualityStatus,
|
|
sample.QualityReason,
|
|
)
|
|
return err
|
|
}
|
|
|
|
func truncateMileageEventTime(value time.Time) time.Time {
|
|
if value.IsZero() {
|
|
return value
|
|
}
|
|
return value.Truncate(time.Second)
|
|
}
|
|
|
|
func NormalizePlatformSourceMileage(ctx context.Context, exec Execer, vin string, statDate string, protocol envelope.Protocol) error {
|
|
if exec == nil {
|
|
panic("stats execer must not be nil")
|
|
}
|
|
if strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" {
|
|
return nil
|
|
}
|
|
_, err := exec.ExecContext(ctx, normalizePlatformSourceMileageInsertSQL, vin, statDate, string(protocol))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = exec.ExecContext(ctx, normalizePlatformSourceMileageDeleteSQL, vin, statDate, string(protocol))
|
|
return err
|
|
}
|
|
|
|
func NormalizePlatformSourceMileageForDate(ctx context.Context, db interface {
|
|
Execer
|
|
Queryer
|
|
}, statDate string, protocol envelope.Protocol) (int, error) {
|
|
if db == nil {
|
|
panic("stats db must not be nil")
|
|
}
|
|
if strings.TrimSpace(statDate) == "" {
|
|
return 0, nil
|
|
}
|
|
rows, err := db.QueryContext(ctx, selectPlatformSourceMileageLegacyVINSQL, statDate, string(protocol))
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var vins []string
|
|
for rows.Next() {
|
|
var vin string
|
|
if err := rows.Scan(&vin); err != nil {
|
|
return 0, err
|
|
}
|
|
if strings.TrimSpace(vin) != "" {
|
|
vins = append(vins, vin)
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
normalized := 0
|
|
for _, vin := range vins {
|
|
if err := NormalizePlatformSourceMileage(ctx, db, vin, statDate, protocol); err != nil {
|
|
return normalized, err
|
|
}
|
|
if err := ProjectDailyMileage(ctx, db, vin, statDate, protocol); err != nil {
|
|
return normalized, err
|
|
}
|
|
normalized++
|
|
}
|
|
return normalized, nil
|
|
}
|
|
|
|
func ShouldNormalizePlatformSourceMileage(sample SourceMileageSample) bool {
|
|
return strings.Contains(sample.SourceKey, platformSourceKeyPrefix)
|
|
}
|
|
|
|
func ProjectDailyMileage(ctx context.Context, exec Execer, vin string, statDate string, protocol envelope.Protocol) error {
|
|
if exec == nil {
|
|
panic("stats execer must not be nil")
|
|
}
|
|
if strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" {
|
|
return nil
|
|
}
|
|
if beginner, ok := exec.(txBeginner); ok {
|
|
tx, err := beginner.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := projectDailyMileageWithExec(ctx, tx, vin, statDate, protocol); err != nil {
|
|
_ = tx.Rollback()
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
return projectDailyMileageWithExec(ctx, exec, vin, statDate, protocol)
|
|
}
|
|
|
|
type txBeginner interface {
|
|
BeginTx(context.Context, *sql.TxOptions) (*sql.Tx, error)
|
|
}
|
|
|
|
func projectDailyMileageWithExec(ctx context.Context, exec Execer, vin string, statDate string, protocol envelope.Protocol) error {
|
|
if _, err := exec.ExecContext(ctx, projectDailyMileageSQL,
|
|
vin,
|
|
statDate,
|
|
string(protocol),
|
|
maxSelectedDailyMileageKM,
|
|
); err != nil {
|
|
return err
|
|
}
|
|
_, err := exec.ExecContext(ctx, markSelectedSourceSQL,
|
|
vin,
|
|
statDate,
|
|
string(protocol),
|
|
maxSelectedDailyMileageKM,
|
|
vin,
|
|
statDate,
|
|
string(protocol),
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = exec.ExecContext(ctx, cleanupProjectedDailyMileageSQL,
|
|
vin,
|
|
statDate,
|
|
string(protocol),
|
|
vin,
|
|
statDate,
|
|
string(protocol),
|
|
)
|
|
return err
|
|
}
|
|
|
|
const upsertSourceStatDateStartSQL = `CAST(CONCAT(VALUES(stat_date), ' 00:00:00') AS DATETIME)`
|
|
|
|
const upsertSourcePreferIncomingFirstSQL = `(
|
|
first_event_time IS NULL
|
|
OR (
|
|
first_event_time < ` + upsertSourceStatDateStartSQL + `
|
|
AND VALUES(first_event_time) < ` + upsertSourceStatDateStartSQL + `
|
|
AND VALUES(first_event_time) >= first_event_time
|
|
)
|
|
OR (
|
|
first_event_time >= ` + upsertSourceStatDateStartSQL + `
|
|
AND VALUES(first_event_time) < ` + upsertSourceStatDateStartSQL + `
|
|
)
|
|
OR (
|
|
first_event_time >= ` + upsertSourceStatDateStartSQL + `
|
|
AND VALUES(first_event_time) >= ` + upsertSourceStatDateStartSQL + `
|
|
AND VALUES(first_event_time) <= first_event_time
|
|
)
|
|
)`
|
|
|
|
const upsertSourceMergedFirstTotalSQL = `CASE
|
|
WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
|
|
THEN VALUES(first_total_mileage_km)
|
|
WHEN ` + upsertSourcePreferIncomingFirstSQL + `
|
|
THEN VALUES(first_total_mileage_km)
|
|
ELSE first_total_mileage_km
|
|
END`
|
|
|
|
const upsertSourceMergedLatestTotalSQL = `CASE
|
|
WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
|
|
THEN VALUES(latest_total_mileage_km)
|
|
WHEN latest_event_time IS NULL OR VALUES(latest_event_time) >= latest_event_time
|
|
THEN VALUES(latest_total_mileage_km)
|
|
ELSE latest_total_mileage_km
|
|
END`
|
|
|
|
const upsertSourceMergedFirstEventSQL = `CASE
|
|
WHEN ` + upsertSourcePreferIncomingFirstSQL + `
|
|
THEN VALUES(first_event_time)
|
|
ELSE first_event_time
|
|
END`
|
|
|
|
const upsertSourceMergedLatestEventSQL = `CASE
|
|
WHEN latest_event_time IS NULL OR VALUES(latest_event_time) >= latest_event_time
|
|
THEN VALUES(latest_event_time)
|
|
ELSE latest_event_time
|
|
END`
|
|
|
|
const upsertSourceMergedDeltaSQL = `(` + upsertSourceMergedLatestTotalSQL + ` - ` + upsertSourceMergedFirstTotalSQL + `)`
|
|
|
|
const upsertSourceMergedDailySQL = `CASE
|
|
WHEN ` + upsertSourceMergedDeltaSQL + ` < 0
|
|
AND ` + upsertSourceMergedDeltaSQL + ` >= -` + maxNegativeMileageJitterKMSQL + `
|
|
THEN 0
|
|
ELSE ` + upsertSourceMergedDeltaSQL + `
|
|
END`
|
|
|
|
const upsertSourceMergedOutsideDailyRangeSQL = `(` + upsertSourceMergedDailySQL + ` < 0 OR ` + upsertSourceMergedDailySQL + ` > ` + upsertSourceMergedMaxMileageSQL + `)`
|
|
|
|
const upsertSourceMergedMissingPreviousBaselineSQL = `(` + upsertSourceMergedFirstEventSQL + ` >= CAST(CONCAT(VALUES(stat_date), ' 00:00:00') AS DATETIME))`
|
|
|
|
const upsertSourceMergedQualityWindowDaysSQL = `GREATEST(1, DATEDIFF(DATE(` + upsertSourceMergedLatestEventSQL + `), DATE(` + upsertSourceMergedFirstEventSQL + `)))`
|
|
|
|
const upsertSourceMergedMaxMileageSQL = `(` + upsertSourceMergedQualityWindowDaysSQL + ` * ` + maxSelectedDailyMileageKMSQL + `)`
|
|
|
|
const normalizePlatformFirstTotalSQL = `CAST(SUBSTRING_INDEX(GROUP_CONCAT(s.first_total_mileage_km ORDER BY s.first_event_time ASC, s.updated_at ASC SEPARATOR ','), ',', 1) AS DECIMAL(18,3))`
|
|
|
|
const normalizePlatformLatestTotalSQL = `CAST(SUBSTRING_INDEX(GROUP_CONCAT(s.latest_total_mileage_km ORDER BY s.latest_event_time DESC, s.updated_at DESC SEPARATOR ','), ',', 1) AS DECIMAL(18,3))`
|
|
|
|
const normalizePlatformDeltaSQL = `(` + normalizePlatformLatestTotalSQL + ` - ` + normalizePlatformFirstTotalSQL + `)`
|
|
|
|
const normalizePlatformDailySQL = `CASE
|
|
WHEN ` + normalizePlatformDeltaSQL + ` < 0
|
|
AND ` + normalizePlatformDeltaSQL + ` >= -` + maxNegativeMileageJitterKMSQL + `
|
|
THEN 0
|
|
ELSE ` + normalizePlatformDeltaSQL + `
|
|
END`
|
|
|
|
const normalizePlatformQualityWindowDaysSQL = `GREATEST(1, DATEDIFF(DATE(MAX(s.latest_event_time)), DATE(MIN(s.first_event_time))))`
|
|
|
|
const normalizePlatformMaxMileageSQL = `(` + normalizePlatformQualityWindowDaysSQL + ` * ` + maxSelectedDailyMileageKMSQL + `)`
|
|
|
|
const normalizePlatformOutsideDailyRangeSQL = `(` + normalizePlatformDailySQL + ` < 0 OR ` + normalizePlatformDailySQL + ` > ` + normalizePlatformMaxMileageSQL + `)`
|
|
|
|
const upsertSourceMileageSQL = `
|
|
INSERT INTO vehicle_daily_mileage_source
|
|
(vin, stat_date, protocol, source_key, source_ip, source_endpoint, phone, device_id, platform_name,
|
|
first_total_mileage_km, latest_total_mileage_km, daily_mileage_km, sample_count,
|
|
first_event_time, latest_event_time, quality_status, quality_reason)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON DUPLICATE KEY UPDATE
|
|
source_ip = VALUES(source_ip),
|
|
source_endpoint = VALUES(source_endpoint),
|
|
phone = VALUES(phone),
|
|
device_id = VALUES(device_id),
|
|
platform_name = COALESCE(NULLIF(TRIM(VALUES(platform_name)), ''), platform_name),
|
|
first_total_mileage_km = ` + upsertSourceMergedFirstTotalSQL + `,
|
|
latest_total_mileage_km = ` + upsertSourceMergedLatestTotalSQL + `,
|
|
daily_mileage_km = ` + upsertSourceMergedDailySQL + `,
|
|
sample_count = sample_count + VALUES(sample_count),
|
|
first_event_time = ` + upsertSourceMergedFirstEventSQL + `,
|
|
latest_event_time = ` + upsertSourceMergedLatestEventSQL + `,
|
|
quality_status = CASE
|
|
WHEN ` + upsertSourceMergedOutsideDailyRangeSQL + ` THEN '` + QualityInvalidDelta + `'
|
|
ELSE '` + QualityOK + `'
|
|
END,
|
|
quality_reason = CASE
|
|
WHEN ` + upsertSourceMergedOutsideDailyRangeSQL + ` THEN 'outside_daily_range'
|
|
WHEN ` + upsertSourceMergedDailySQL + ` = 0 AND ` + upsertSourceMergedDeltaSQL + ` < 0 THEN 'negative_jitter_clamped'
|
|
WHEN ` + upsertSourceMergedMissingPreviousBaselineSQL + ` THEN '` + QualityReasonCurrentDayFirst + `'
|
|
ELSE '` + QualityReasonHistorical + `'
|
|
END,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
`
|
|
|
|
const normalizePlatformSourceMileageInsertSQL = `
|
|
INSERT INTO vehicle_daily_mileage_source
|
|
(vin, stat_date, protocol, source_key, source_ip, source_endpoint, phone, device_id, platform_name,
|
|
first_total_mileage_km, latest_total_mileage_km, daily_mileage_km, sample_count,
|
|
first_event_time, latest_event_time, quality_status, quality_reason)
|
|
SELECT
|
|
s.vin,
|
|
s.stat_date,
|
|
s.protocol,
|
|
CONCAT(s.protocol, ':', COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), '')), '` + platformSourceKeyPrefix + `', COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code)) AS stable_source_key,
|
|
SUBSTRING_INDEX(GROUP_CONCAT(s.source_ip ORDER BY s.latest_event_time DESC, s.updated_at DESC SEPARATOR ','), ',', 1) AS source_ip,
|
|
SUBSTRING_INDEX(GROUP_CONCAT(COALESCE(s.source_endpoint, '') ORDER BY s.latest_event_time DESC, s.updated_at DESC SEPARATOR ','), ',', 1) AS source_endpoint,
|
|
SUBSTRING_INDEX(GROUP_CONCAT(COALESCE(s.phone, '') ORDER BY s.latest_event_time DESC, s.updated_at DESC SEPARATOR ','), ',', 1) AS phone,
|
|
SUBSTRING_INDEX(GROUP_CONCAT(COALESCE(s.device_id, '') ORDER BY s.latest_event_time DESC, s.updated_at DESC SEPARATOR ','), ',', 1) AS device_id,
|
|
COALESCE(NULLIF(TRIM(SUBSTRING_INDEX(GROUP_CONCAT(COALESCE(NULLIF(TRIM(s.platform_name), ''), ds.platform_name, vi.platform_name, '') ORDER BY s.latest_event_time DESC, s.updated_at DESC SEPARATOR ','), ',', 1)), ''), MAX(TRIM(COALESCE(ds.platform_name, vi.platform_name)))) AS platform_name,
|
|
` + normalizePlatformFirstTotalSQL + ` AS first_total_mileage_km,
|
|
` + normalizePlatformLatestTotalSQL + ` AS latest_total_mileage_km,
|
|
` + normalizePlatformDailySQL + ` AS daily_mileage_km,
|
|
SUM(s.sample_count) AS sample_count,
|
|
MIN(s.first_event_time) AS first_event_time,
|
|
MAX(s.latest_event_time) AS latest_event_time,
|
|
CASE
|
|
WHEN ` + normalizePlatformOutsideDailyRangeSQL + ` THEN '` + QualityInvalidDelta + `'
|
|
ELSE '` + QualityOK + `'
|
|
END AS quality_status,
|
|
CASE
|
|
WHEN ` + normalizePlatformOutsideDailyRangeSQL + ` THEN 'outside_daily_range'
|
|
WHEN ` + normalizePlatformDailySQL + ` = 0 AND ` + normalizePlatformDeltaSQL + ` < 0 THEN 'negative_jitter_clamped'
|
|
WHEN MIN(s.first_event_time) < CAST(CONCAT(s.stat_date, ' 00:00:00') AS DATETIME)
|
|
THEN '` + QualityReasonHistorical + `'
|
|
ELSE '` + QualityReasonCurrentDayFirst + `'
|
|
END AS quality_reason
|
|
FROM vehicle_daily_mileage_source s
|
|
LEFT JOIN vehicle_data_source ds
|
|
ON ds.protocol = s.protocol AND ds.source_ip = s.source_ip
|
|
AND ds.enabled = 1
|
|
AND ds.source_kind = 'PLATFORM'
|
|
AND ds.source_code IS NOT NULL
|
|
AND TRIM(ds.source_code) <> ''
|
|
LEFT JOIN (
|
|
SELECT
|
|
identifier_value AS phone,
|
|
MIN(TRIM(source_code)) AS source_code,
|
|
MIN(COALESCE(NULLIF(TRIM(oem), ''), NULLIF(TRIM(source_code), ''))) AS platform_name
|
|
FROM vehicle_identifier
|
|
WHERE protocol = 'JT808'
|
|
AND identifier_type = 'JT808_PHONE'
|
|
AND enabled = 1
|
|
AND source_code IS NOT NULL
|
|
AND TRIM(source_code) <> ''
|
|
GROUP BY identifier_value
|
|
HAVING COUNT(DISTINCT TRIM(source_code)) = 1
|
|
) vi
|
|
ON s.protocol = 'JT808' AND vi.phone = TRIM(s.phone)
|
|
WHERE s.vin = ?
|
|
AND s.stat_date = ?
|
|
AND s.protocol = ?
|
|
AND s.quality_status IN ('` + QualityOK + `', '` + QualityNoPreviousBaseline + `')
|
|
AND s.first_total_mileage_km IS NOT NULL
|
|
AND s.latest_total_mileage_km IS NOT NULL
|
|
AND COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code) IS NOT NULL
|
|
AND COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), '')) IS NOT NULL
|
|
AND s.source_key <> CONCAT(s.protocol, ':', COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), '')), '` + platformSourceKeyPrefix + `', COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code))
|
|
GROUP BY s.vin, s.stat_date, s.protocol, stable_source_key
|
|
ON DUPLICATE KEY UPDATE
|
|
source_ip = CASE
|
|
WHEN latest_event_time IS NULL OR VALUES(latest_event_time) >= latest_event_time THEN VALUES(source_ip)
|
|
ELSE source_ip
|
|
END,
|
|
source_endpoint = CASE
|
|
WHEN latest_event_time IS NULL OR VALUES(latest_event_time) >= latest_event_time THEN VALUES(source_endpoint)
|
|
ELSE source_endpoint
|
|
END,
|
|
phone = COALESCE(NULLIF(TRIM(VALUES(phone)), ''), phone),
|
|
device_id = COALESCE(NULLIF(TRIM(VALUES(device_id)), ''), device_id),
|
|
platform_name = COALESCE(NULLIF(TRIM(VALUES(platform_name)), ''), platform_name),
|
|
first_total_mileage_km = ` + upsertSourceMergedFirstTotalSQL + `,
|
|
latest_total_mileage_km = ` + upsertSourceMergedLatestTotalSQL + `,
|
|
daily_mileage_km = ` + upsertSourceMergedDailySQL + `,
|
|
sample_count = sample_count + VALUES(sample_count),
|
|
first_event_time = CASE
|
|
WHEN first_event_time IS NULL OR VALUES(first_event_time) <= first_event_time THEN VALUES(first_event_time)
|
|
ELSE first_event_time
|
|
END,
|
|
latest_event_time = CASE
|
|
WHEN latest_event_time IS NULL OR VALUES(latest_event_time) >= latest_event_time THEN VALUES(latest_event_time)
|
|
ELSE latest_event_time
|
|
END,
|
|
quality_status = CASE
|
|
WHEN ` + upsertSourceMergedOutsideDailyRangeSQL + ` THEN '` + QualityInvalidDelta + `'
|
|
ELSE '` + QualityOK + `'
|
|
END,
|
|
quality_reason = CASE
|
|
WHEN ` + upsertSourceMergedOutsideDailyRangeSQL + ` THEN 'outside_daily_range'
|
|
WHEN ` + upsertSourceMergedDailySQL + ` = 0 AND ` + upsertSourceMergedDeltaSQL + ` < 0 THEN 'negative_jitter_clamped'
|
|
WHEN ` + upsertSourceMergedMissingPreviousBaselineSQL + ` THEN '` + QualityReasonCurrentDayFirst + `'
|
|
ELSE '` + QualityReasonHistorical + `'
|
|
END,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
`
|
|
|
|
const normalizePlatformSourceMileageDeleteSQL = `
|
|
DELETE s
|
|
FROM vehicle_daily_mileage_source s
|
|
LEFT JOIN vehicle_data_source ds
|
|
ON ds.protocol = s.protocol AND ds.source_ip = s.source_ip
|
|
AND ds.enabled = 1
|
|
AND ds.source_kind = 'PLATFORM'
|
|
AND ds.source_code IS NOT NULL
|
|
AND TRIM(ds.source_code) <> ''
|
|
LEFT JOIN (
|
|
SELECT
|
|
identifier_value AS phone,
|
|
MIN(TRIM(source_code)) AS source_code
|
|
FROM vehicle_identifier
|
|
WHERE protocol = 'JT808'
|
|
AND identifier_type = 'JT808_PHONE'
|
|
AND enabled = 1
|
|
AND source_code IS NOT NULL
|
|
AND TRIM(source_code) <> ''
|
|
GROUP BY identifier_value
|
|
HAVING COUNT(DISTINCT TRIM(source_code)) = 1
|
|
) vi
|
|
ON s.protocol = 'JT808' AND vi.phone = TRIM(s.phone)
|
|
WHERE s.vin = ?
|
|
AND s.stat_date = ?
|
|
AND s.protocol = ?
|
|
AND s.quality_status IN ('` + QualityOK + `', '` + QualityNoPreviousBaseline + `')
|
|
AND s.first_total_mileage_km IS NOT NULL
|
|
AND s.latest_total_mileage_km IS NOT NULL
|
|
AND COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code) IS NOT NULL
|
|
AND COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), '')) IS NOT NULL
|
|
AND s.source_key <> CONCAT(s.protocol, ':', COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), '')), '` + platformSourceKeyPrefix + `', COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code))
|
|
`
|
|
|
|
const selectPlatformSourceMileageLegacyVINSQL = `
|
|
SELECT DISTINCT s.vin
|
|
FROM vehicle_daily_mileage_source s
|
|
LEFT JOIN vehicle_data_source ds
|
|
ON ds.protocol = s.protocol AND ds.source_ip = s.source_ip
|
|
AND ds.enabled = 1
|
|
AND ds.source_kind = 'PLATFORM'
|
|
AND ds.source_code IS NOT NULL
|
|
AND TRIM(ds.source_code) <> ''
|
|
LEFT JOIN (
|
|
SELECT
|
|
identifier_value AS phone,
|
|
MIN(TRIM(source_code)) AS source_code
|
|
FROM vehicle_identifier
|
|
WHERE protocol = 'JT808'
|
|
AND identifier_type = 'JT808_PHONE'
|
|
AND enabled = 1
|
|
AND source_code IS NOT NULL
|
|
AND TRIM(source_code) <> ''
|
|
GROUP BY identifier_value
|
|
HAVING COUNT(DISTINCT TRIM(source_code)) = 1
|
|
) vi
|
|
ON s.protocol = 'JT808' AND vi.phone = TRIM(s.phone)
|
|
WHERE s.stat_date = ?
|
|
AND s.protocol = ?
|
|
AND s.quality_status IN ('` + QualityOK + `', '` + QualityNoPreviousBaseline + `')
|
|
AND COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code) IS NOT NULL
|
|
AND COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), '')) IS NOT NULL
|
|
AND s.source_key <> CONCAT(s.protocol, ':', COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), '')), '` + platformSourceKeyPrefix + `', COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code))
|
|
ORDER BY s.vin
|
|
`
|
|
|
|
const projectDailyMileageSQL = `
|
|
INSERT INTO vehicle_daily_mileage
|
|
(vin, stat_date, protocol, source_id, daily_mileage_km, latest_total_mileage_km)
|
|
SELECT
|
|
s.vin,
|
|
s.stat_date,
|
|
s.protocol,
|
|
ds.id,
|
|
s.daily_mileage_km,
|
|
s.latest_total_mileage_km
|
|
FROM vehicle_daily_mileage_source s
|
|
LEFT JOIN vehicle_data_source ds
|
|
ON ds.protocol = s.protocol AND ds.source_ip = s.source_ip
|
|
WHERE s.vin = ?
|
|
AND s.stat_date = ?
|
|
AND s.protocol = ?
|
|
AND s.quality_status = '` + QualityOK + `'
|
|
AND s.daily_mileage_km BETWEEN 0 AND (? * GREATEST(1, DATEDIFF(DATE(s.latest_event_time), DATE(s.first_event_time))))
|
|
AND (ds.id IS NULL OR ds.enabled = 1 OR (
|
|
COALESCE(NULLIF(TRIM(ds.source_kind), ''), 'UNKNOWN') = 'UNKNOWN'
|
|
AND (ds.source_code IS NULL OR TRIM(ds.source_code) = '')
|
|
AND (ds.platform_name IS NULL OR TRIM(ds.platform_name) = '')
|
|
))
|
|
ORDER BY CASE COALESCE(NULLIF(NULLIF(TRIM(ds.source_kind), ''), 'UNKNOWN'), CASE WHEN s.source_key LIKE '%` + platformSourceKeyPrefix + `%' THEN 'PLATFORM' WHEN s.source_key LIKE '%` + directSourceKeySuffix + `' THEN 'DIRECT' ELSE 'UNKNOWN' END)
|
|
WHEN 'PLATFORM' THEN 0
|
|
WHEN 'DIRECT' THEN 1
|
|
WHEN 'UNKNOWN' THEN 2
|
|
ELSE 3
|
|
END,
|
|
COALESCE(ds.trust_priority, 100000),
|
|
s.sample_count DESC,
|
|
s.latest_event_time DESC,
|
|
s.source_key ASC
|
|
LIMIT 1
|
|
ON DUPLICATE KEY UPDATE
|
|
source_id = VALUES(source_id),
|
|
daily_mileage_km = VALUES(daily_mileage_km),
|
|
latest_total_mileage_km = VALUES(latest_total_mileage_km),
|
|
updated_at = CURRENT_TIMESTAMP
|
|
`
|
|
|
|
const markSelectedSourceSQL = `
|
|
UPDATE vehicle_daily_mileage_source s
|
|
LEFT JOIN (
|
|
SELECT
|
|
s2.source_key,
|
|
s2.vin,
|
|
s2.stat_date,
|
|
s2.protocol
|
|
FROM vehicle_daily_mileage_source s2
|
|
LEFT JOIN vehicle_data_source ds
|
|
ON ds.protocol = s2.protocol AND ds.source_ip = s2.source_ip
|
|
WHERE s2.vin = ?
|
|
AND s2.stat_date = ?
|
|
AND s2.protocol = ?
|
|
AND s2.quality_status = '` + QualityOK + `'
|
|
AND s2.daily_mileage_km BETWEEN 0 AND (? * GREATEST(1, DATEDIFF(DATE(s2.latest_event_time), DATE(s2.first_event_time))))
|
|
AND (ds.id IS NULL OR ds.enabled = 1 OR (
|
|
COALESCE(NULLIF(TRIM(ds.source_kind), ''), 'UNKNOWN') = 'UNKNOWN'
|
|
AND (ds.source_code IS NULL OR TRIM(ds.source_code) = '')
|
|
AND (ds.platform_name IS NULL OR TRIM(ds.platform_name) = '')
|
|
))
|
|
ORDER BY CASE COALESCE(NULLIF(NULLIF(TRIM(ds.source_kind), ''), 'UNKNOWN'), CASE WHEN s2.source_key LIKE '%` + platformSourceKeyPrefix + `%' THEN 'PLATFORM' WHEN s2.source_key LIKE '%` + directSourceKeySuffix + `' THEN 'DIRECT' ELSE 'UNKNOWN' END)
|
|
WHEN 'PLATFORM' THEN 0
|
|
WHEN 'DIRECT' THEN 1
|
|
WHEN 'UNKNOWN' THEN 2
|
|
ELSE 3
|
|
END,
|
|
COALESCE(ds.trust_priority, 100000),
|
|
s2.sample_count DESC,
|
|
s2.latest_event_time DESC,
|
|
s2.source_key ASC
|
|
LIMIT 1
|
|
) selected_source
|
|
ON selected_source.source_key = s.source_key
|
|
AND selected_source.vin = s.vin
|
|
AND selected_source.stat_date = s.stat_date
|
|
AND selected_source.protocol = s.protocol
|
|
SET s.is_selected = CASE WHEN selected_source.source_key IS NULL THEN 0 ELSE 1 END
|
|
WHERE s.vin = ? AND s.stat_date = ? AND s.protocol = ?
|
|
`
|
|
|
|
const cleanupProjectedDailyMileageSQL = `
|
|
DELETE FROM vehicle_daily_mileage
|
|
WHERE vin = ? AND stat_date = ? AND protocol = ?
|
|
AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM vehicle_daily_mileage_source
|
|
WHERE vin = ? AND stat_date = ? AND protocol = ? AND is_selected = 1
|
|
)
|
|
`
|
|
|
|
type sourceBaseline struct {
|
|
LatestTotalKM float64
|
|
LatestEventTime time.Time
|
|
QualityReason string
|
|
}
|
|
|
|
func lookupPreviousSourceBaseline(ctx context.Context, query Queryer, vin string, statDate string, protocol envelope.Protocol, sourceKey string) (sourceBaseline, bool, error) {
|
|
if query == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" || strings.TrimSpace(sourceKey) == "" {
|
|
return sourceBaseline{}, false, nil
|
|
}
|
|
rows, err := query.QueryContext(ctx, previousSourceBaselineSQL, vin, statDate, string(protocol), sourceKey)
|
|
if err != nil {
|
|
return sourceBaseline{}, false, err
|
|
}
|
|
defer rows.Close()
|
|
if !rows.Next() {
|
|
if err := rows.Err(); err != nil {
|
|
return sourceBaseline{}, false, err
|
|
}
|
|
return sourceBaseline{}, false, nil
|
|
}
|
|
var latestTotal sql.NullFloat64
|
|
var latestEvent sql.NullTime
|
|
if err := rows.Scan(&latestTotal, &latestEvent); err != nil {
|
|
return sourceBaseline{}, false, err
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return sourceBaseline{}, false, err
|
|
}
|
|
return sourceBaseline{
|
|
LatestTotalKM: latestTotal.Float64,
|
|
LatestEventTime: latestEvent.Time,
|
|
QualityReason: QualityReasonHistorical,
|
|
}, latestTotal.Valid, nil
|
|
}
|
|
|
|
// LookupLatestSourceBaselineBefore returns the nearest durable odometer for the
|
|
// same VIN, protocol and source before statDate. Missing calendar days are
|
|
// skipped automatically.
|
|
func LookupLatestSourceBaselineBefore(ctx context.Context, query Queryer, vin string, statDate string, protocol envelope.Protocol, sourceKey string) (float64, time.Time, bool, error) {
|
|
baseline, found, err := lookupPreviousSourceBaseline(ctx, query, vin, statDate, protocol, sourceKey)
|
|
return baseline.LatestTotalKM, baseline.LatestEventTime, found, err
|
|
}
|
|
|
|
const previousSourceBaselineSQL = `
|
|
SELECT latest_total_mileage_km, latest_event_time
|
|
FROM vehicle_daily_mileage_source
|
|
WHERE vin = ?
|
|
AND stat_date < ?
|
|
AND protocol = ?
|
|
AND source_key = ?
|
|
AND latest_total_mileage_km IS NOT NULL
|
|
AND latest_total_mileage_km > 0
|
|
AND latest_event_time IS NOT NULL
|
|
AND latest_event_time >= CAST(CONCAT(stat_date, ' 00:00:00') AS DATETIME)
|
|
AND latest_event_time < DATE_ADD(CAST(CONCAT(stat_date, ' 00:00:00') AS DATETIME), INTERVAL 1 DAY)
|
|
ORDER BY stat_date DESC, latest_event_time DESC
|
|
LIMIT 1
|
|
`
|