lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill/main.go

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
}

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.Count,
		)
	}
	sqlText := `INSERT INTO vehicle_daily_mileage
  (vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, 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),
  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]float64{}
	for _, protocol := range cfg.Protocols {
		lastByProtocolDate[protocol] = map[string]map[string]float64{}
		for _, date := range dates {
			rows, err := queryDailyLastRows(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, latest := range current {
				first, ok := previous[vin]
				if !ok || latest < first {
					continue
				}
				key := vin + "|" + date + "|" + string(protocol)
				aggregates[key] = &metricAgg{
					VIN:      vin,
					Date:     date,
					Protocol: protocol,
					FirstKM:  first,
					LatestKM: latest,
					Count:    1,
				}
			}
		}
	}
	return aggregates, nil
}

func queryDailyLastRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string]float64, 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, LAST(parsed_json)
FROM %s.raw_frames
WHERE %s
GROUP BY vin`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
	rows, err := db.QueryContext(ctx, sqlText)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[string]float64{}
	for rows.Next() {
		var vin string
		var parsedJSON string
		if err := rows.Scan(&vin, &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
		}
		out[strings.TrimSpace(vin)] = samples[0].TotalMileageKM
	}
	return out, rows.Err()
}

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