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
	loc              *time.Location
	mu               sync.Mutex
	lastTotalMileage map[string]float64
}

type MetricSample struct {
	VIN            string
	Protocol       envelope.Protocol
	StatDate       string
	TotalMileageKM float64
}

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)
	}
	return &Writer{exec: exec, loc: loc, lastTotalMileage: map[string]float64{}}
}

func (w *Writer) EnsureSchema(ctx context.Context) error {
	if _, err := w.exec.ExecContext(ctx, DailyMileageTableSQL); err != nil {
		return err
	}
	return nil
}

func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
	samples, err := SamplesFromEnvelope(env, w.loc)
	if err != nil {
		return err
	}
	for _, sample := range samples {
		if w.seenSameMileage(sample) {
			continue
		}
		if _, err := w.exec.ExecContext(ctx, upsertDailyMileageSQL,
			sample.VIN,
			sample.StatDate,
			string(sample.Protocol),
			sample.TotalMileageKM,
			sample.TotalMileageKM); err != nil {
			return err
		}
	}
	return nil
}

func (w *Writer) seenSameMileage(sample MetricSample) bool {
	prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol)
	key := prefix + sample.StatDate
	w.mu.Lock()
	defer w.mu.Unlock()
	if last, ok := w.lastTotalMileage[key]; ok && last == sample.TotalMileageKM {
		return true
	}
	for existing := range w.lastTotalMileage {
		if strings.HasPrefix(existing, prefix) && existing != key {
			delete(w.lastTotalMileage, existing)
		}
	}
	w.lastTotalMileage[key] = sample.TotalMileageKM
	return false
}

func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) {
	vin := strings.TrimSpace(env.VIN)
	if vin == "" {
		return nil, nil
	}
	totalMileage, ok := floatField(env, envelope.FieldTotalMileageKM)
	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,
	}}, nil
}

const upsertDailyMileageSQL = `
INSERT INTO vehicle_daily_mileage
  (vin, stat_date, protocol, daily_mileage_km,
   first_total_mileage_km, latest_total_mileage_km, sample_count)
VALUES (?, ?, ?, 0, ?, ?, 1)
ON DUPLICATE KEY UPDATE
  first_total_mileage_km = CASE
    WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
      THEN VALUES(first_total_mileage_km)
    ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
  END,
  latest_total_mileage_km = CASE
    WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
      THEN VALUES(latest_total_mileage_km)
    ELSE GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
  END,
  daily_mileage_km = GREATEST(
    CASE
      WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
        THEN VALUES(latest_total_mileage_km)
      ELSE latest_total_mileage_km
    END,
    VALUES(latest_total_mileage_km)
  ) - CASE
    WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
      THEN VALUES(first_total_mileage_km)
    ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
  END,
  sample_count = sample_count + 1,
  updated_at = CURRENT_TIMESTAMP
`

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