package main import ( "context" "database/sql" "encoding/json" "fmt" "log/slog" "os" "regexp" "strconv" "strings" "time" _ "github.com/go-sql-driver/mysql" _ "github.com/taosdata/driver-go/v3/taosWS" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/history" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats" ) type config struct { MySQLDSN string TDengineDriver string TDengineDSN string TDengineDatabase string DateFrom string DateTo string Protocols []envelope.Protocol Method string Limit int DryRun bool Reset bool Debug bool ProgressEvery int64 Location *time.Location } type rawFrameRow struct { Protocol envelope.Protocol VIN string EventID string MessageID string EventTimeMS int64 ReceivedAtMS int64 ParsedJSON string } type metricAgg struct { VIN string Date string Protocol envelope.Protocol FirstKM float64 LatestKM float64 Count int64 SourceKey string Phone string SourceEndpoint string } type dailySourceLast struct { VIN string SourceKey string Phone string SourceEndpoint string TS time.Time TotalKM float64 } func main() { if err := loadEnvFiles(env("BACKFILL_ENV_FILES", os.Getenv("BACKFILL_ENV_FILE"))); err != nil { fail("load env file", err) } cfg, err := loadConfig() if err != nil { fail("load config", err) } ctx := context.Background() td, err := sql.Open(cfg.TDengineDriver, cfg.TDengineDSN) if err != nil { fail("open tdengine", err) } defer td.Close() if err := td.PingContext(ctx); err != nil { fail("ping tdengine", err) } mysqlDB, err := sql.Open("mysql", cfg.MySQLDSN) if err != nil { fail("open mysql", err) } defer mysqlDB.Close() if err := mysqlDB.PingContext(ctx); err != nil { fail("ping mysql", err) } schemaWriter := stats.NewWriter(mysqlDB, cfg.Location) if err := schemaWriter.EnsureSchema(ctx); err != nil { fail("ensure schema", err) } if cfg.Reset && !cfg.DryRun { deleted, err := resetStats(ctx, mysqlDB, cfg) if err != nil { fail("reset stats", err) } slog.Info("reset stats rows", "deleted", deleted) } if cfg.Method == "last_diff" { aggregates, err := buildLastDiffAggregates(ctx, td, cfg) if err != nil { fail("build last-diff aggregates", err) } var written int64 if !cfg.DryRun { written, err = writeAggregates(ctx, mysqlDB, aggregates, 500) if err != nil { fail("write aggregates", err) } } slog.Info("stats backfill complete", "method", cfg.Method, "dry_run", cfg.DryRun, "aggregates", len(aggregates), "written", written) return } rows, err := queryRawFrames(ctx, td, cfg) if err != nil { fail("query raw frames", err) } defer rows.Close() aggregates := map[string]*metricAgg{} var scanned, chunked, parsed, sampled, written, skipped int64 for rows.Next() { scanned++ row, err := scanRawFrame(rows) if err != nil { fail("scan raw frame", err) } text := strings.TrimSpace(row.ParsedJSON) if isChunkManifest(text) { chunked++ text, err = readChunkedPayload(ctx, td, cfg.TDengineDatabase, row.EventID, "parsed_fields") if err != nil { slog.Warn("skip chunked parsed fields", "event_id", row.EventID, "error", err) skipped++ continue } } fields := fieldsForStats(row.Protocol, row.VIN, text) if cfg.Debug && scanned <= 5 { slog.Info("debug raw frame", "protocol", row.Protocol, "vin", row.VIN, "message_id", row.MessageID, "parsed_len", len(text), "mileage_keys", mileageKeys(row.Protocol), "extracted_fields", fields) } if len(fields) == 0 { skipped++ continue } parsed++ env := envelope.FrameEnvelope{ EventID: row.EventID, Protocol: row.Protocol, MessageID: row.MessageID, VIN: row.VIN, EventTimeMS: row.EventTimeMS, ReceivedAtMS: row.ReceivedAtMS, Fields: fields, ParsedFields: fields, ParseStatus: envelope.ParseOK, } samples, err := stats.SamplesFromEnvelope(env, cfg.Location) if err != nil { slog.Warn("skip invalid sample", "event_id", row.EventID, "error", err) skipped++ continue } if len(samples) == 0 { skipped++ continue } sampled += int64(len(samples)) addSamples(aggregates, samples) if cfg.ProgressEvery > 0 && scanned%cfg.ProgressEvery == 0 { slog.Info("stats backfill progress", "scanned", scanned, "sampled", sampled, "aggregates", len(aggregates), "skipped", skipped) } } if err := rows.Err(); err != nil { fail("iterate raw frames", err) } if !cfg.DryRun { var err error written, err = writeAggregates(ctx, mysqlDB, aggregates, 500) if err != nil { fail("write aggregates", err) } } slog.Info("stats backfill complete", "dry_run", cfg.DryRun, "scanned", scanned, "parsed", parsed, "chunked", chunked, "sampled", sampled, "aggregates", len(aggregates), "written", written, "skipped", skipped) } func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample) { for _, sample := range samples { key := sample.VIN + "|" + sample.StatDate + "|" + string(sample.Protocol) agg, ok := aggregates[key] if !ok { aggregates[key] = &metricAgg{ VIN: sample.VIN, Date: sample.StatDate, Protocol: sample.Protocol, FirstKM: sample.TotalMileageKM, LatestKM: sample.TotalMileageKM, Count: 1, } continue } if sample.TotalMileageKM < agg.FirstKM { agg.FirstKM = sample.TotalMileageKM } if sample.TotalMileageKM > agg.LatestKM { agg.LatestKM = sample.TotalMileageKM } agg.Count++ } } func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*metricAgg, batchSize int) (int64, error) { if len(aggregates) == 0 { return 0, nil } if batchSize <= 0 { batchSize = 500 } rows := make([]*metricAgg, 0, len(aggregates)) for _, agg := range aggregates { rows = append(rows, agg) } var written int64 for start := 0; start < len(rows); start += batchSize { end := start + batchSize if end > len(rows) { end = len(rows) } if err := writeAggregateBatch(ctx, db, rows[start:end]); err != nil { return written, err } written += int64(end - start) } return written, nil } func writeAggregateBatch(ctx context.Context, db *sql.DB, rows []*metricAgg) error { placeholders := make([]string, 0, len(rows)) args := make([]any, 0, len(rows)*7) for _, row := range rows { daily := row.LatestKM - row.FirstKM if daily < 0 { daily = 0 } placeholders = append(placeholders, "(?,?,?,?,?,?,?,?,?,?)") args = append(args, row.VIN, row.Date, string(row.Protocol), daily, row.FirstKM, row.LatestKM, row.SourceKey, row.Phone, row.SourceEndpoint, row.Count, ) } sqlText := `INSERT INTO vehicle_daily_mileage (vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count) VALUES ` + strings.Join(placeholders, ",") + ` ON DUPLICATE KEY UPDATE daily_mileage_km = VALUES(daily_mileage_km), first_total_mileage_km = VALUES(first_total_mileage_km), latest_total_mileage_km = VALUES(latest_total_mileage_km), trusted_source_key = VALUES(trusted_source_key), trusted_phone = VALUES(trusted_phone), trusted_source_endpoint = VALUES(trusted_source_endpoint), sample_count = VALUES(sample_count), updated_at = CURRENT_TIMESTAMP` _, err := db.ExecContext(ctx, sqlText, args...) return err } func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[string]*metricAgg, error) { dates, err := dateRangeWithPrevious(cfg.DateFrom, cfg.DateTo) if err != nil { return nil, err } lastByProtocolDate := map[envelope.Protocol]map[string]map[string][]dailySourceLast{} for _, protocol := range cfg.Protocols { lastByProtocolDate[protocol] = map[string]map[string][]dailySourceLast{} for _, date := range dates { rows, err := queryDailyLastSourceRows(ctx, db, cfg, protocol, date) if err != nil { return nil, err } lastByProtocolDate[protocol][date] = rows slog.Info("daily last loaded", "protocol", protocol, "date", date, "vehicles", len(rows)) } } targetDates, err := dateRange(cfg.DateFrom, cfg.DateTo) if err != nil { return nil, err } aggregates := map[string]*metricAgg{} for _, protocol := range cfg.Protocols { for _, date := range targetDates { prevDate := previousDate(date) current := lastByProtocolDate[protocol][date] previous := lastByProtocolDate[protocol][prevDate] for vin, currentSources := range current { chosen, ok := chooseTrustedSource(currentSources, previous[vin]) if !ok { continue } key := vin + "|" + date + "|" + string(protocol) aggregates[key] = &metricAgg{ VIN: vin, Date: date, Protocol: protocol, FirstKM: chosen.previous.TotalKM, LatestKM: chosen.current.TotalKM, Count: 1, SourceKey: chosen.current.SourceKey, Phone: chosen.current.Phone, SourceEndpoint: chosen.current.SourceEndpoint, } } } } return aggregates, nil } type trustedChoice struct { current dailySourceLast previous dailySourceLast } const maxTrustedDailyMileageKM = 1000 func chooseTrustedSource(current []dailySourceLast, previous []dailySourceLast) (trustedChoice, bool) { previousBySource := map[string]dailySourceLast{} for _, row := range previous { previousBySource[row.SourceKey] = row } var chosen trustedChoice var chosenDelta float64 for _, currentRow := range current { previousRow, ok := previousBySource[currentRow.SourceKey] if !ok { continue } delta := currentRow.TotalKM - previousRow.TotalKM if delta < 0 || delta > maxTrustedDailyMileageKM { continue } if chosen.current.SourceKey == "" || delta < chosenDelta { chosen = trustedChoice{current: currentRow, previous: previousRow} chosenDelta = delta } } return chosen, chosen.current.SourceKey != "" } func queryDailyLastSourceRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string][]dailySourceLast, error) { where := []string{ fmt.Sprintf("ts >= '%s 00:00:00'", quote(date)), fmt.Sprintf("ts < '%s 00:00:00'", quote(nextDate(date))), "parse_status = 'OK'", "vin IS NOT NULL", "vin <> ''", fmt.Sprintf("protocol = '%s'", quote(string(protocol))), realtimeMileageFramePredicate(), } sqlText := fmt.Sprintf(`SELECT vin, phone, source_endpoint, LAST(ts), LAST(parsed_json) FROM %s.raw_frames WHERE %s GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) rows, err := db.QueryContext(ctx, sqlText) if err != nil { return nil, err } defer rows.Close() latestBySource := map[string]dailySourceLast{} for rows.Next() { var vin string var phone string var sourceEndpoint string var ts time.Time var parsedJSON string if err := rows.Scan(&vin, &phone, &sourceEndpoint, &ts, &parsedJSON); err != nil { return nil, err } fields := fieldsForStats(protocol, vin, parsedJSON) env := envelope.FrameEnvelope{ Protocol: protocol, VIN: strings.TrimSpace(vin), EventTimeMS: time.Now().UnixMilli(), ReceivedAtMS: time.Now().UnixMilli(), Fields: fields, ParsedFields: fields, ParseStatus: envelope.ParseOK, } samples, err := stats.SamplesFromEnvelope(env, cfg.Location) if err != nil || len(samples) == 0 { continue } sourceKey := normalizedSourceKey(phone, sourceEndpoint) if sourceKey == "" { continue } row := dailySourceLast{ VIN: strings.TrimSpace(vin), SourceKey: sourceKey, Phone: strings.TrimSpace(phone), SourceEndpoint: strings.TrimSpace(sourceEndpoint), TS: ts, TotalKM: samples[0].TotalMileageKM, } key := row.VIN + "|" + row.SourceKey if existing, ok := latestBySource[key]; !ok || row.TS.After(existing.TS) { latestBySource[key] = row } } if err := rows.Err(); err != nil { return nil, err } out := map[string][]dailySourceLast{} for _, row := range latestBySource { out[row.VIN] = append(out[row.VIN], row) } return out, nil } func normalizedSourceKey(phone string, endpoint string) string { var parts []string if phone = strings.TrimSpace(phone); phone != "" { parts = append(parts, phone) } if host := endpointHost(endpoint); host != "" { parts = append(parts, host) } return strings.Join(parts, "@") } func endpointHost(endpoint string) string { endpoint = strings.TrimSpace(endpoint) if endpoint == "" { return "" } if host, _, ok := strings.Cut(endpoint, ":"); ok { return strings.TrimSpace(host) } return endpoint } func fieldsForStats(protocol envelope.Protocol, vin string, text string) map[string]any { fields := map[string]any{} text = strings.TrimSpace(text) if text == "" { return fields } parsed := map[string]any{} if err := json.Unmarshal([]byte(text), &parsed); err == nil && len(parsed) > 0 { if flattened, _, ok := realtime.ParsedFieldsForEnvelope(envelope.FrameEnvelope{ Protocol: protocol, VIN: vin, Parsed: parsed, }); ok { fields = flattened } else { fields = parsed } } for _, key := range mileageKeys(protocol) { if _, exists := fields[key]; exists { continue } value, ok := extractJSONStringField(text, key) if !ok { continue } fields[key] = value } return fields } func mileageKeys(protocol envelope.Protocol) []string { switch protocol { case envelope.ProtocolGB32960: return []string{"gb32960.vehicle.total_mileage_km"} case envelope.ProtocolJT808: return []string{"jt808.location.total_mileage_km"} case envelope.ProtocolYutongMQTT: return []string{"yutong_mqtt.data.total_mileage", "yutong_mqtt.root.data.total_mileage"} default: return nil } } func extractJSONStringField(text string, key string) (string, bool) { pattern := regexp.QuoteMeta(`"`+key+`"`) + `\s*:\s*"?([^",}]+)"?` match := regexp.MustCompile(pattern).FindStringSubmatch(text) if len(match) != 2 { return "", false } return strings.TrimSpace(match[1]), true } func loadConfig() (config, error) { loc, err := time.LoadLocation(env("LOCAL_TZ", "Asia/Shanghai")) if err != nil { loc = time.FixedZone("Asia/Shanghai", 8*3600) } dateTo := env("BACKFILL_DATE_TO", time.Now().In(loc).Format("2006-01-02")) dateFrom := env("BACKFILL_DATE_FROM", dateTo) protocols, err := parseProtocols(env("BACKFILL_PROTOCOLS", "GB32960,JT808,YUTONG_MQTT")) if err != nil { return config{}, err } return config{ MySQLDSN: env("MYSQL_DSN", ""), TDengineDriver: env("TDENGINE_DRIVER", "taosWS"), TDengineDSN: env("TDENGINE_DSN", ""), TDengineDatabase: env("TDENGINE_DATABASE", history.DefaultDatabase), DateFrom: dateFrom, DateTo: dateTo, Protocols: protocols, Method: env("BACKFILL_METHOD", "scan"), Limit: envInt("BACKFILL_LIMIT", 0), DryRun: envBool("BACKFILL_DRY_RUN", true), Reset: envBool("BACKFILL_RESET", false), Debug: envBool("BACKFILL_DEBUG", false), ProgressEvery: int64(envInt("BACKFILL_PROGRESS_EVERY", 100000)), Location: loc, }, nil } func queryRawFrames(ctx context.Context, db *sql.DB, cfg config) (*sql.Rows, error) { where := []string{ fmt.Sprintf("ts >= '%s 00:00:00'", quote(cfg.DateFrom)), fmt.Sprintf("ts < '%s 00:00:00'", quote(nextDate(cfg.DateTo))), "parse_status = 'OK'", "vin IS NOT NULL", "vin <> ''", } if len(cfg.Protocols) > 0 { quoted := make([]string, 0, len(cfg.Protocols)) for _, protocol := range cfg.Protocols { quoted = append(quoted, "'"+quote(string(protocol))+"'") } where = append(where, "protocol IN ("+strings.Join(quoted, ",")+")") } where = append(where, realtimeMileageFramePredicate()) sqlText := fmt.Sprintf(`SELECT protocol, vin, event_id, message_id, event_time, received_at, parsed_json FROM %s.raw_frames WHERE %s ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) if cfg.Limit > 0 { sqlText += fmt.Sprintf(" LIMIT %d", cfg.Limit) } slog.Info("query raw frames", "date_from", cfg.DateFrom, "date_to", cfg.DateTo, "protocols", cfg.Protocols, "limit", cfg.Limit) return db.QueryContext(ctx, sqlText) } func realtimeMileageFramePredicate() string { return `( (protocol = 'GB32960' AND message_id IN (2,3)) OR (protocol = 'JT808' AND message_id = 512) OR (protocol = 'YUTONG_MQTT') )` } func scanRawFrame(rows *sql.Rows) (rawFrameRow, error) { var protocol, vin, eventID, parsed string var messageID int64 var eventTime, receivedAt time.Time if err := rows.Scan(&protocol, &vin, &eventID, &messageID, &eventTime, &receivedAt, &parsed); err != nil { return rawFrameRow{}, err } return rawFrameRow{ Protocol: envelope.Protocol(strings.TrimSpace(protocol)), VIN: strings.TrimSpace(vin), EventID: strings.TrimSpace(eventID), MessageID: fmt.Sprintf("0x%04X", messageID), EventTimeMS: eventTime.UnixMilli(), ReceivedAtMS: receivedAt.UnixMilli(), ParsedJSON: parsed, }, nil } func isChunkManifest(text string) bool { if !strings.Contains(text, "chunked") || !strings.Contains(text, "payload_kind") { return false } var manifest struct { Chunked bool `json:"chunked"` } return json.Unmarshal([]byte(text), &manifest) == nil && manifest.Chunked } func readChunkedPayload(ctx context.Context, db *sql.DB, database string, eventID string, kind string) (string, error) { sqlText := fmt.Sprintf(`SELECT chunk_text FROM %s.raw_frame_payload_chunks WHERE event_id = '%s' AND payload_kind = '%s' ORDER BY chunk_index ASC`, ident(database), quote(eventID), quote(kind)) rows, err := db.QueryContext(ctx, sqlText) if err != nil { return "", err } defer rows.Close() var b strings.Builder for rows.Next() { var chunk string if err := rows.Scan(&chunk); err != nil { return "", err } b.WriteString(chunk) } if err := rows.Err(); err != nil { return "", err } if b.Len() == 0 { return "", fmt.Errorf("no chunks found") } return b.String(), nil } func resetStats(ctx context.Context, db *sql.DB, cfg config) (int64, error) { clauses := []string{"stat_date >= ?", "stat_date <= ?"} args := []any{cfg.DateFrom, cfg.DateTo} if len(cfg.Protocols) > 0 { placeholders := make([]string, 0, len(cfg.Protocols)) for _, protocol := range cfg.Protocols { placeholders = append(placeholders, "?") args = append(args, string(protocol)) } clauses = append(clauses, "protocol IN ("+strings.Join(placeholders, ",")+")") } result, err := db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage WHERE "+strings.Join(clauses, " AND "), args...) if err != nil { return 0, err } return result.RowsAffected() } func parseProtocols(value string) ([]envelope.Protocol, error) { var protocols []envelope.Protocol for _, part := range strings.Split(value, ",") { part = strings.ToUpper(strings.TrimSpace(part)) if part == "" { continue } switch envelope.Protocol(part) { case envelope.ProtocolGB32960, envelope.ProtocolJT808, envelope.ProtocolYutongMQTT: protocols = append(protocols, envelope.Protocol(part)) default: return nil, fmt.Errorf("unsupported protocol %q", part) } } return protocols, nil } func loadEnvFiles(paths string) error { for _, path := range strings.Split(paths, ",") { if err := loadEnvFile(path); err != nil { return err } } return nil } func loadEnvFile(path string) error { path = strings.TrimSpace(path) if path == "" { return nil } data, err := os.ReadFile(path) if err != nil { return err } for _, line := range strings.Split(string(data), "\n") { line = strings.TrimSpace(line) if line == "" || strings.HasPrefix(line, "#") || !strings.Contains(line, "=") { continue } key, value, _ := strings.Cut(line, "=") key = strings.TrimSpace(key) value = strings.Trim(strings.TrimSpace(value), "\"'") if key != "" && os.Getenv(key) == "" { _ = os.Setenv(key, value) } } return nil } func env(key string, fallback string) string { value := strings.TrimSpace(os.Getenv(key)) if value == "" { return fallback } return value } func envInt(key string, fallback int) int { value := strings.TrimSpace(os.Getenv(key)) if value == "" { return fallback } parsed, err := strconv.Atoi(value) if err != nil { return fallback } return parsed } func envBool(key string, fallback bool) bool { value := strings.ToLower(strings.TrimSpace(os.Getenv(key))) if value == "" { return fallback } return value == "1" || value == "true" || value == "yes" || value == "y" } func nextDate(date string) string { parsed, err := time.Parse("2006-01-02", date) if err != nil { return date } return parsed.AddDate(0, 0, 1).Format("2006-01-02") } func previousDate(date string) string { parsed, err := time.Parse("2006-01-02", date) if err != nil { return date } return parsed.AddDate(0, 0, -1).Format("2006-01-02") } func dateRangeWithPrevious(from string, to string) ([]string, error) { dates, err := dateRange(from, to) if err != nil { return nil, err } return append([]string{previousDate(from)}, dates...), nil } func dateRange(from string, to string) ([]string, error) { start, err := time.Parse("2006-01-02", from) if err != nil { return nil, err } end, err := time.Parse("2006-01-02", to) if err != nil { return nil, err } if end.Before(start) { return nil, fmt.Errorf("dateTo before dateFrom") } var dates []string for day := start; !day.After(end); day = day.AddDate(0, 0, 1) { dates = append(dates, day.Format("2006-01-02")) } return dates, nil } func ident(value string) string { value = strings.TrimSpace(strings.Trim(value, "`")) if value == "" { return history.DefaultDatabase } return value } func quote(value string) string { return strings.ReplaceAll(value, "'", "''") } func fail(stage string, err error) { slog.Error(stage, "error", err) os.Exit(1) }