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

package stats

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

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

const (
	MetricDailyMileageKM      = "daily_mileage_km"
	MetricDailyTotalMileageKM = "daily_total_mileage_km"
)

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

type Writer struct {
	exec Execer
	loc  *time.Location
}

type MetricSample struct {
	VIN            string
	Protocol       envelope.Protocol
	StatDate       string
	MetricKey      string
	MetricValue    float64
	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}
}

func (w *Writer) EnsureSchema(ctx context.Context) error {
	_, err := w.exec.ExecContext(ctx, DailyMetricTableSQL)
	return err
}

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 _, err := w.exec.ExecContext(ctx, upsertDailyMetricSQL,
			sample.VIN,
			sample.StatDate,
			string(sample.Protocol),
			sample.MetricKey,
			sample.MetricValue,
			sample.TotalMileageKM,
			sample.TotalMileageKM); err != nil {
			return err
		}
	}
	return nil
}

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,
			MetricKey:      MetricDailyMileageKM,
			MetricValue:    0,
			TotalMileageKM: totalMileage,
		},
		{
			VIN:            vin,
			Protocol:       env.Protocol,
			StatDate:       statDate,
			MetricKey:      MetricDailyTotalMileageKM,
			MetricValue:    totalMileage,
			TotalMileageKM: totalMileage,
		},
	}, nil
}

const upsertDailyMetricSQL = `
INSERT INTO vehicle_daily_metric
  (vin, stat_date, protocol, metric_key, metric_value, metric_unit,
   first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method)
VALUES (?, ?, ?, ?, ?, 'km', ?, ?, 1, 'TOTAL_MILEAGE_DIFF')
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,
  metric_value = CASE
    WHEN metric_key = 'daily_mileage_km'
      THEN 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
    WHEN metric_key = 'daily_total_mileage_km'
      THEN 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)
           )
    ELSE VALUES(metric_value)
  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
	}
}