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
	Phone          string
	DeviceID       string
	SourceEndpoint 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
	DeviceID        string
	SourceEndpoint  string
	FirstEventTime  time.Time
	LatestEventTime time.Time
	QualityStatus   string
	QualityReason   string
}

type dailySourceLast struct {
	VIN            string
	SourceKey      string
	Phone          string
	DeviceID       string
	SourceEndpoint string
	FirstTS        time.Time
	TS             time.Time
	FirstTotalKM   float64
	TotalKM        float64
	RawSampleCount 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,
			Phone:          row.Phone,
			DeviceID:       row.DeviceID,
			SourceEndpoint: row.SourceEndpoint,
			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) + "|" + sample.SourceKey
		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,
				SourceKey:       sample.SourceKey,
				Phone:           sample.Phone,
				DeviceID:        sample.DeviceID,
				SourceEndpoint:  sample.SourceEndpoint,
				FirstEventTime:  sample.EventTime,
				LatestEventTime: sample.EventTime,
				QualityStatus:   stats.QualityOK,
				QualityReason:   "current_day_first_sample",
			}
			continue
		}
		if agg.FirstEventTime.IsZero() || sample.EventTime.Before(agg.FirstEventTime) {
			agg.FirstKM = sample.TotalMileageKM
			agg.FirstEventTime = sample.EventTime
		}
		if sample.EventTime.After(agg.LatestEventTime) {
			agg.LatestKM = sample.TotalMileageKM
			agg.LatestEventTime = sample.EventTime
			agg.SourceEndpoint = sample.SourceEndpoint
			if strings.TrimSpace(sample.Phone) != "" {
				agg.Phone = sample.Phone
			}
			if strings.TrimSpace(sample.DeviceID) != "" {
				agg.DeviceID = sample.DeviceID
			}
		}
		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
	}
	var written int64
	clearedTargets := map[string]struct{}{}
	for _, agg := range aggregates {
		identity := stats.SourceIdentity{
			Protocol:       agg.Protocol,
			SourceIP:       stats.NormalizeSourceIP(agg.SourceEndpoint),
			SourceEndpoint: agg.SourceEndpoint,
		}
		if strings.TrimSpace(identity.SourceIP) == "" {
			continue
		}
		target := agg.VIN + "|" + agg.Date + "|" + string(agg.Protocol)
		if _, ok := clearedTargets[target]; !ok {
			if err := clearBackfillTargetMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
				return written, err
			}
			clearedTargets[target] = struct{}{}
		}
		if err := stats.UpsertDataSource(ctx, db, identity, agg.LatestEventTime); err != nil {
			return written, err
		}
		dailyKM := agg.LatestKM - agg.FirstKM
		candidate := stats.SourceMileageSample{
			VIN:             agg.VIN,
			StatDate:        agg.Date,
			Protocol:        agg.Protocol,
			SourceKey:       agg.SourceKey,
			SourceIP:        identity.SourceIP,
			SourceEndpoint:  agg.SourceEndpoint,
			Phone:           agg.Phone,
			DeviceID:        agg.DeviceID,
			FirstTotalKM:    agg.FirstKM,
			LatestTotalKM:   agg.LatestKM,
			DailyKM:         dailyKM,
			SampleCount:     agg.Count,
			FirstEventTime:  agg.FirstEventTime,
			LatestEventTime: agg.LatestEventTime,
			QualityStatus:   agg.QualityStatus,
			QualityReason:   agg.QualityReason,
		}
		if candidate.QualityStatus == "" {
			candidate.QualityStatus = stats.QualityOK
		}
		if candidate.QualityReason == "" {
			candidate.QualityReason = "same_source_previous_day"
		}
		if candidate.QualityStatus == stats.QualityOK && (candidate.DailyKM < 0 || candidate.DailyKM > maxTrustedDailyMileageKM) {
			candidate.QualityStatus = stats.QualityInvalidDelta
			candidate.QualityReason = "outside_daily_range"
		}
		if err := stats.UpsertSourceMileage(ctx, db, candidate); err != nil {
			return written, err
		}
		if err := stats.ProjectDailyMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
			return written, err
		}
		written++
	}
	return written, nil
}

func clearBackfillTargetMileage(ctx context.Context, db *sql.DB, vin string, statDate string, protocol envelope.Protocol) error {
	if db == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" || strings.TrimSpace(string(protocol)) == "" {
		return nil
	}
	_, err := db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage_source WHERE vin = ? AND stat_date = ? AND protocol = ?", vin, statDate, string(protocol))
	if err != nil {
		return err
	}
	_, err = db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage WHERE vin = ? AND stat_date = ? AND protocol = ?", vin, statDate, string(protocol))
	return err
}

func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[string]*metricAgg, error) {
	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 {
			current, err := queryDailyLastSourceRows(ctx, db, cfg, protocol, date)
			if err != nil {
				return nil, err
			}
			previous, err := queryPreviousLastSourceRows(ctx, db, cfg, protocol, date)
			if err != nil {
				return nil, err
			}
			slog.Info("daily last loaded", "protocol", protocol, "date", date, "vehicles", len(current), "previousVehicles", len(previous))
			for vin, currentSources := range current {
				previousBySource := map[string]dailySourceLast{}
				for _, previousRow := range previous[vin] {
					previousBySource[previousRow.SourceKey] = previousRow
				}
				for _, currentRow := range currentSources {
					key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey
					previousRow, hasPrevious := previousBySource[currentRow.SourceKey]
					aggregates[key] = aggregateFromDailySource(date, protocol, currentRow, previousRow, hasPrevious)
				}
			}
		}
	}
	return aggregates, nil
}

func aggregateFromDailySource(date string, protocol envelope.Protocol, current dailySourceLast, previous dailySourceLast, hasPrevious bool) *metricAgg {
	firstKM := current.FirstTotalKM
	firstEventTime := current.FirstTS
	qualityReason := "current_day_first_sample"
	if hasPrevious {
		firstKM = previous.TotalKM
		firstEventTime = previous.TS
		qualityReason = "historical_source_baseline"
	}
	if firstEventTime.IsZero() {
		firstEventTime = current.TS
	}
	count := current.RawSampleCount
	if count <= 0 {
		count = 1
	}
	return &metricAgg{
		VIN:             current.VIN,
		Date:            date,
		Protocol:        protocol,
		FirstKM:         firstKM,
		LatestKM:        current.TotalKM,
		Count:           count,
		SourceKey:       current.SourceKey,
		Phone:           current.Phone,
		DeviceID:        current.DeviceID,
		SourceEndpoint:  current.SourceEndpoint,
		FirstEventTime:  firstEventTime,
		LatestEventTime: current.TS,
		QualityStatus:   stats.QualityOK,
		QualityReason:   qualityReason,
	}
}

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(),
	}
	if predicate := mileageBearingFramePredicate(protocol); predicate != "" {
		where = append(where, predicate)
	}
	sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, FIRST(ts), FIRST(parsed_json), LAST(ts), LAST(parsed_json), COUNT(*)
FROM %s.raw_frames
WHERE %s
GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
	return querySourceRows(ctx, db, cfg, protocol, sqlText, true)
}

func queryPreviousLastSourceRows(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)),
		"parse_status = 'OK'",
		"vin IS NOT NULL",
		"vin <> ''",
		fmt.Sprintf("protocol = '%s'", quote(string(protocol))),
		realtimeMileageFramePredicate(),
	}
	if predicate := mileageBearingFramePredicate(protocol); predicate != "" {
		where = append(where, predicate)
	}
	sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, LAST(ts), LAST(parsed_json), COUNT(*)
FROM %s.raw_frames
WHERE %s
GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
	return querySourceRows(ctx, db, cfg, protocol, sqlText, false)
}

func querySourceRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, sqlText string, includeFirst bool) (map[string][]dailySourceLast, error) {
	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 deviceID string
		var sourceEndpoint string
		var firstTS time.Time
		var firstParsedJSON string
		var ts time.Time
		var parsedJSON string
		var rawSampleCount int64
		if includeFirst {
			if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &firstTS, &firstParsedJSON, &ts, &parsedJSON, &rawSampleCount); err != nil {
				return nil, err
			}
		} else {
			if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &ts, &parsedJSON, &rawSampleCount); err != nil {
				return nil, err
			}
			firstTS = ts
			firstParsedJSON = parsedJSON
		}
		latestTotalKM, ok := mileageFromParsed(protocol, vin, parsedJSON, ts, cfg.Location)
		if !ok {
			continue
		}
		firstTotalKM := latestTotalKM
		if includeFirst {
			if parsedFirst, ok := mileageFromParsed(protocol, vin, firstParsedJSON, firstTS, cfg.Location); ok {
				firstTotalKM = parsedFirst
			}
		}
		sourceKey := normalizedSourceKey(string(protocol), phone, deviceID, sourceEndpoint)
		if sourceKey == "" {
			continue
		}
		row := dailySourceLast{
			VIN:            strings.TrimSpace(vin),
			SourceKey:      sourceKey,
			Phone:          strings.TrimSpace(phone),
			DeviceID:       strings.TrimSpace(deviceID),
			SourceEndpoint: strings.TrimSpace(sourceEndpoint),
			FirstTS:        firstTS.In(cfg.Location),
			TS:             ts,
			FirstTotalKM:   firstTotalKM,
			TotalKM:        latestTotalKM,
			RawSampleCount: rawSampleCount,
		}
		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 mileageFromParsed(protocol envelope.Protocol, vin string, parsedJSON string, eventTime time.Time, loc *time.Location) (float64, bool) {
	fields := fieldsForStats(protocol, vin, parsedJSON)
	env := envelope.FrameEnvelope{
		Protocol:     protocol,
		VIN:          strings.TrimSpace(vin),
		EventTimeMS:  eventTime.UnixMilli(),
		ReceivedAtMS: eventTime.UnixMilli(),
		Fields:       fields,
		ParsedFields: fields,
		ParseStatus:  envelope.ParseOK,
	}
	samples, err := stats.SamplesFromEnvelope(env, loc)
	if err != nil || len(samples) == 0 {
		return 0, false
	}
	return samples[0].TotalMileageKM, true
}

func normalizedSourceKey(protocol string, phone string, deviceID string, endpoint string) string {
	sourceIP := stats.NormalizeSourceIP(endpoint)
	return stats.SourceKey(envelope.Protocol(protocol), phone, deviceID, sourceIP)
}

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", "last_diff"),
		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, phone, device_id, source_endpoint, 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 mileageBearingFramePredicate(protocol envelope.Protocol) string {
	keys := mileageKeys(protocol)
	if len(keys) == 0 {
		return ""
	}
	conditions := make([]string, 0, len(keys))
	for _, key := range keys {
		conditions = append(conditions, fmt.Sprintf("parsed_json LIKE '%%%s%%'", quote(key)))
	}
	return "(" + strings.Join(conditions, " OR ") + ")"
}

func scanRawFrame(rows *sql.Rows) (rawFrameRow, error) {
	var protocol, vin, phone, deviceID, sourceEndpoint, eventID, parsed string
	var messageID int64
	var eventTime, receivedAt time.Time
	if err := rows.Scan(&protocol, &vin, &phone, &deviceID, &sourceEndpoint, &eventID, &messageID, &eventTime, &receivedAt, &parsed); err != nil {
		return rawFrameRow{}, err
	}
	return rawFrameRow{
		Protocol:       envelope.Protocol(strings.TrimSpace(protocol)),
		VIN:            strings.TrimSpace(vin),
		Phone:          strings.TrimSpace(phone),
		DeviceID:       strings.TrimSpace(deviceID),
		SourceEndpoint: strings.TrimSpace(sourceEndpoint),
		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, ",")+")")
	}
	var deleted int64
	for _, table := range []string{"vehicle_daily_mileage_source", "vehicle_daily_mileage"} {
		result, err := db.ExecContext(ctx, "DELETE FROM "+table+" WHERE "+strings.Join(clauses, " AND "), args...)
		if err != nil {
			return deleted, err
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return deleted, err
		}
		deleted += affected
	}
	return deleted, nil
}

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