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

package stats

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

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

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

type Writer struct {
	exec                Execer
	query               Queryer
	loc                 *time.Location
	sourceTouchInterval time.Duration
	mu                  sync.Mutex
	lastTotalMileage    map[string]float64
	lastSourceSeen      map[string]time.Time
	baselineCache       map[string]sourceBaselineCacheEntry
}

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

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,
		lastTotalMileage:    map[string]float64{},
		lastSourceSeen:      map[string]time.Time{},
		baselineCache:       map[string]sourceBaselineCacheEntry{},
	}
	if query, ok := exec.(Queryer); ok {
		writer.query = query
	}
	return writer
}

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

func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
	identity, hasSource := NewSourceIdentity(env.Protocol, env.SourceEndpoint)
	if hasSource {
		seenAt := w.sourceSeenAt(env)
		if w.shouldTouchSource(identity, seenAt) {
			if err := UpsertDataSource(ctx, w.exec, identity, seenAt); err != nil {
				return err
			}
			w.markSourceTouched(identity, seenAt)
		}
	}
	samples, err := SamplesFromEnvelope(env, w.loc)
	if err != nil {
		return err
	}
	for _, sample := range samples {
		if w.seenSameMileage(sample) {
			continue
		}
		if !hasSource {
			continue
		}
		candidate := SourceMileageSampleFromMetric(sample, identity)
		if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil {
			return err
		}
		if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil {
			return err
		}
		if err := ProjectDailyMileage(ctx, w.exec, sample.VIN, sample.StatDate, sample.Protocol); err != nil {
			return err
		}
		w.markMileageWritten(sample)
	}
	return nil
}

func (w *Writer) sourceSeenAt(env envelope.FrameEnvelope) time.Time {
	eventMS := env.EventTimeMS
	if eventMS <= 0 {
		eventMS = env.ReceivedAtMS
	}
	if eventMS > 0 {
		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()
	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.mu.Unlock()
}

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()
	for existing := range w.lastTotalMileage {
		if strings.HasPrefix(existing, prefix) && existing != key {
			delete(w.lastTotalMileage, existing)
		}
	}
	for existing := range w.baselineCache {
		if strings.HasPrefix(existing, prefix) && existing != key {
			delete(w.baselineCache, existing)
		}
	}
	w.lastTotalMileage[key] = sample.TotalMileageKM
}

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 {
		candidate.QualityStatus = QualityNoPreviousBaseline
		candidate.QualityReason = "missing_previous_source"
		return nil
	}
	candidate.FirstTotalKM = baseline.LatestTotalKM
	candidate.FirstEventTime = baseline.LatestEventTime
	candidate.DailyKM = candidate.LatestTotalKM - baseline.LatestTotalKM
	candidate.QualityStatus = QualityOK
	candidate.QualityReason = "same_source_previous_day"
	if candidate.DailyKM < 0 || candidate.DailyKM > maxSelectedDailyMileageKM {
		candidate.QualityStatus = QualityInvalidDelta
		candidate.QualityReason = "outside_daily_range"
	}
	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,
	})
	w.mu.Lock()
	if cached, ok := w.baselineCache[cacheKey]; ok {
		w.mu.Unlock()
		return cached.baseline, cached.found, nil
	}
	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 {
		baseline, found, err = lookupCurrentSourceBaseline(ctx, w.query, candidate.VIN, candidate.StatDate, candidate.Protocol, candidate.SourceKey)
		if err != nil {
			return sourceBaseline{}, false, err
		}
	}

	if found {
		w.mu.Lock()
		w.baselineCache[cacheKey] = sourceBaselineCacheEntry{baseline: baseline, found: true}
		w.mu.Unlock()
	}
	return baseline, found, nil
}

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
}

type sourceBaselineCacheEntry struct {
	baseline sourceBaseline
	found    bool
}

func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) {
	vin := strings.TrimSpace(env.VIN)
	if vin == "" {
		return nil, nil
	}
	totalMileage, ok := totalMileageKMFromEnvelope(env)
	if !ok {
		return nil, nil
	}
	if totalMileage <= 0 {
		return nil, nil
	}
	if loc == nil {
		loc = time.FixedZone("Asia/Shanghai", 8*3600)
	}
	eventMS := env.EventTimeMS
	if eventMS <= 0 {
		eventMS = env.ReceivedAtMS
	}
	if eventMS <= 0 {
		return nil, errors.New("event or received time is required")
	}
	statDate := time.UnixMilli(eventMS).In(loc).Format("2006-01-02")
	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),
	}}, nil
}

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

func isDuplicateColumnError(err error) bool {
	if err == nil {
		return false
	}
	text := strings.ToLower(err.Error())
	return strings.Contains(text, "duplicate column") || strings.Contains(text, "1060")
}

type mileageFieldMapping struct {
	key   string
	scale float64
}

func totalMileageKMFromEnvelope(env envelope.FrameEnvelope) (float64, bool) {
	if value, ok := floatField(env, envelope.FieldTotalMileageKM); ok {
		return value, true
	}
	for _, mapping := range mileageMappingsByProtocol(env.Protocol) {
		value, ok := floatField(env, mapping.key)
		if !ok {
			continue
		}
		return value * mapping.scale, true
	}
	return 0, false
}

func mileageMappingsByProtocol(protocol envelope.Protocol) []mileageFieldMapping {
	switch protocol {
	case envelope.ProtocolGB32960:
		return []mileageFieldMapping{{key: "gb32960.vehicle.total_mileage_km", scale: 1}}
	case envelope.ProtocolJT808:
		return []mileageFieldMapping{{key: "jt808.location.total_mileage_km", scale: 1}}
	case envelope.ProtocolYutongMQTT:
		return []mileageFieldMapping{
			{key: "yutong_mqtt.data.total_mileage", scale: 0.001},
			{key: "yutong_mqtt.root.data.total_mileage", scale: 0.001},
		}
	default:
		return nil
	}
}

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
	}
}