shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(go): simplify daily mileage storage

Browse Source
This commit is contained in:
lingniu
2026-07-02 20:01:03 +08:00
parent abc1b9870a
commit 593dc08870
7 changed files with 74 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
go/vehicle-gateway/internal/stats/daily_metric.go
View File

@@ -11,11 +11,6 @@ import (
"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)
}
@@ -30,8 +25,6 @@ type MetricSample struct {
VIN string
Protocol envelope.Protocol
StatDate string
MetricKey string
MetricValue float64
TotalMileageKM float64
}
@@ -46,12 +39,9 @@ func NewWriter(exec Execer, loc *time.Location) *Writer {
}
func (w *Writer) EnsureSchema(ctx context.Context) error {
if _, err := w.exec.ExecContext(ctx, DailyMetricTableSQL); err != nil {
if _, err := w.exec.ExecContext(ctx, DailyMileageTableSQL); err != nil {
return err
}
if db, ok := w.exec.(metadataDB); ok {
return migrateDailyMetricTable(ctx, db)
}
return nil
}
@@ -61,13 +51,11 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
return err
}
for _, sample := range samples {
if _, err := w.exec.ExecContext(ctx, upsertDailyMetricSQL,
if _, err := w.exec.ExecContext(ctx, upsertDailyMileageSQL,
sample.VehicleKey,
sample.VIN,
sample.StatDate,
string(sample.Protocol),
sample.MetricKey,
sample.MetricValue,
sample.TotalMileageKM,
sample.TotalMileageKM); err != nil {
return err
@@ -100,33 +88,20 @@ func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]Metr
return nil, errors.New("event or received time is required")
}
statDate := time.UnixMilli(eventMS).In(loc).Format("2006-01-02")
return []MetricSample{
{
VehicleKey: vehicleKey,
VIN: vin,
Protocol: env.Protocol,
StatDate: statDate,
MetricKey: MetricDailyMileageKM,
MetricValue: 0,
TotalMileageKM: totalMileage,
},
{
VehicleKey: vehicleKey,
VIN: vin,
Protocol: env.Protocol,
StatDate: statDate,
MetricKey: MetricDailyTotalMileageKM,
MetricValue: totalMileage,
TotalMileageKM: totalMileage,
},
}, nil
return []MetricSample{{
VehicleKey: vehicleKey,
VIN: vin,
Protocol: env.Protocol,
StatDate: statDate,
TotalMileageKM: totalMileage,
}}, nil
}
const upsertDailyMetricSQL = `
INSERT INTO vehicle_daily_metric
(vehicle_key, 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')
const upsertDailyMileageSQL = `
INSERT INTO vehicle_daily_mileage
(vehicle_key, vin, stat_date, protocol, daily_mileage_km,
first_total_mileage_km, latest_total_mileage_km, sample_count)
VALUES (?, ?, ?, ?, 0, ?, ?, 1)
ON DUPLICATE KEY UPDATE
vin = IF(VALUES(vin) <> '', VALUES(vin), vin),
first_total_mileage_km = CASE
@@ -139,112 +114,22 @@ ON DUPLICATE KEY UPDATE
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)
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
`
type metadataDB interface {
Execer
QueryRowContext(context.Context, string, ...any) *sql.Row
}
func migrateDailyMetricTable(ctx context.Context, db metadataDB) error {
columnExists, err := informationSchemaExists(ctx, db, `
SELECT COUNT(*)
FROM information_schema.columns
WHERE table_schema = DATABASE()
AND table_name = 'vehicle_daily_metric'
AND column_name = 'vehicle_key'`)
if err != nil {
return err
}
if !columnExists {
if _, err := db.ExecContext(ctx, `ALTER TABLE vehicle_daily_metric ADD COLUMN vehicle_key VARCHAR(96) NOT NULL DEFAULT '' AFTER id`); err != nil {
return err
}
}
if _, err := db.ExecContext(ctx, `UPDATE vehicle_daily_metric SET vehicle_key = vin WHERE vehicle_key = ''`); err != nil {
return err
}
vehicleIndexExists, err := informationSchemaExists(ctx, db, `
SELECT COUNT(*)
FROM information_schema.statistics
WHERE table_schema = DATABASE()
AND table_name = 'vehicle_daily_metric'
AND index_name = 'uk_daily_metric_vehicle'`)
if err != nil {
return err
}
if !vehicleIndexExists {
if _, err := db.ExecContext(ctx, `ALTER TABLE vehicle_daily_metric ADD UNIQUE KEY uk_daily_metric_vehicle (vehicle_key, stat_date, protocol, metric_key)`); err != nil {
return err
}
}
oldIndexExists, err := informationSchemaExists(ctx, db, `
SELECT COUNT(*)
FROM information_schema.statistics
WHERE table_schema = DATABASE()
AND table_name = 'vehicle_daily_metric'
AND index_name = 'uk_daily_metric'`)
if err != nil {
return err
}
if oldIndexExists {
if _, err := db.ExecContext(ctx, `ALTER TABLE vehicle_daily_metric DROP INDEX uk_daily_metric`); err != nil {
return err
}
}
vinIndexExists, err := informationSchemaExists(ctx, db, `
SELECT COUNT(*)
FROM information_schema.statistics
WHERE table_schema = DATABASE()
AND table_name = 'vehicle_daily_metric'
AND index_name = 'idx_vin'`)
if err != nil {
return err
}
if !vinIndexExists {
if _, err := db.ExecContext(ctx, `ALTER TABLE vehicle_daily_metric ADD KEY idx_vin (vin)`); err != nil {
return err
}
}
return nil
}
func informationSchemaExists(ctx context.Context, db metadataDB, query string) (bool, error) {
var count int
if err := db.QueryRowContext(ctx, query).Scan(&count); err != nil {
return false, err
}
return count > 0, nil
}
func floatField(env envelope.FrameEnvelope, key string) (float64, bool) {
if env.Fields == nil {
return 0, false