lingniu-vehicle-ingest/vehicle-data-platform/apps/api/internal/openplatform/realtime_live.go

package openplatform

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"math"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"time"
)

// GB32960 live freshness is an API policy, not an assertion about the device's
// negotiated reporting interval. No supported protocol currently has a verified
// full-vehicle percentage/capacity contract.
const realtimeHydrogenStaleSeconds int64 = 300

type RealtimeHydrogenData struct {
	RemainingHydrogenKg             *float64 `json:"remainingHydrogenKg"`
	RemainingHydrogenPercent        *float64 `json:"remainingHydrogenPercent"`
	HydrogenRecordTime              *string  `json:"hydrogenRecordTime"`
	HydrogenDataStatus              string   `json:"hydrogenDataStatus"`
	RemainingHydrogenKgStatus       string   `json:"remainingHydrogenKgStatus"`
	RemainingHydrogenPercentStatus  string   `json:"remainingHydrogenPercentStatus"`
	HydrogenValueSource             *string  `json:"hydrogenValueSource"`
	HydrogenSourceProtocol          *string  `json:"hydrogenSourceProtocol"`
	HydrogenStaleAfterSeconds       *int64   `json:"hydrogenStaleAfterSeconds"`
	HydrogenExpectedIntervalSeconds *int64   `json:"hydrogenExpectedIntervalSeconds"`
}

func missingRealtimeHydrogen(protocol string) RealtimeHydrogenData {
	status := "UNSUPPORTED"
	if protocol == "" || protocol == "GB32960" {
		status = "MISSING"
	}
	result := RealtimeHydrogenData{HydrogenDataStatus: status, RemainingHydrogenKgStatus: status, RemainingHydrogenPercentStatus: "UNSUPPORTED"}
	if protocol == "GB32960" {
		threshold := realtimeHydrogenStaleSeconds
		result.HydrogenStaleAfterSeconds = &threshold
		result.HydrogenSourceProtocol = &protocol
	}
	return result
}

func applyRealtimeLiveData(item *RealtimeVehicleResult, point RealtimeVehiclePoint, location *time.Location) {
	item.RealtimeHydrogenData = point.LiveHydrogen
	if item.HydrogenDataStatus == "" {
		item.RealtimeHydrogenData = missingRealtimeHydrogen(point.Protocol)
	}
	if point.GPSFixStatus != "" {
		item.GPSFixStatus = point.GPSFixStatus
	}
	if point.CoordinateSystem != "" {
		item.CoordinateSystem = point.CoordinateSystem
	}
	if !point.LocationObservedAt.IsZero() {
		at := point.LocationObservedAt.In(location).Format(time.RFC3339Nano)
		item.LocationRecordTime = &at
	}
}

func realtimeHydrogenFromFrame(parsed string, observedAt, now time.Time) RealtimeHydrogenData {
	result := missingRealtimeHydrogen("GB32960")
	var fields map[string]any
	if json.Unmarshal([]byte(parsed), &fields) != nil {
		return result
	}
	for _, key := range hydrogenMassFields {
		value, exists := fields[key]
		if !exists {
			continue
		}
		// A malformed or explicit protocol invalid value must never fall through to
		// an alias retaining a different reading.
		mass, valid := numericValue(value)
		if !observedAt.IsZero() {
			at := observedAt.In(time.FixedZone("Asia/Shanghai", 8*60*60)).Format(time.RFC3339Nano)
			result.HydrogenRecordTime = &at
		}
		if !valid || math.IsNaN(mass) || math.IsInf(mass, 0) || mass < 0 || mass > 200 || observedAt.IsZero() || observedAt.After(now.Add(time.Minute)) {
			result.HydrogenDataStatus, result.RemainingHydrogenKgStatus = "INVALID", "INVALID"
			return result
		}
		mass = round3(mass)
		result.RemainingHydrogenKg = &mass
		source := "REPORTED"
		result.HydrogenValueSource = &source
		result.HydrogenDataStatus, result.RemainingHydrogenKgStatus = "PARTIAL", "NORMAL"
		if now.Sub(observedAt) > time.Duration(realtimeHydrogenStaleSeconds)*time.Second {
			result.HydrogenDataStatus, result.RemainingHydrogenKgStatus = "STALE", "STALE"
		}
		return result
	}
	return result
}

// GPS flags are read only from the exact raw frame which produced the chosen
// location. A snapshot's JSON is merge-patched and cannot prove field freshness.
func realtimeGPSFromFrame(protocol, parsed string) (string, string) {
	fix, coordinate := "UNKNOWN", "UNKNOWN"
	var fields map[string]any
	if json.Unmarshal([]byte(parsed), &fields) != nil {
		return fix, coordinate
	}
	switch protocol {
	case "GB32960":
		if flag, ok := numericValue(fields["gb32960.position.position_status"]); ok && flag >= 0 && flag <= 7 && math.Trunc(flag) == flag {
			fix = "FIXED"
			if int(flag)&1 != 0 {
				fix = "NO_FIX"
			}
		}
		if code, exists := fields["gb32960.position.coordinate_system"]; exists {
			if value, ok := numericValue(code); ok {
				switch value {
				case 1:
					coordinate = "WGS84"
				case 2:
					coordinate = "GCJ02"
				}
			}
		}
	case "JT808":
		if flag, ok := numericValue(fields["jt808.location.status_flag"]); ok && flag >= 0 && flag <= math.MaxUint32 && math.Trunc(flag) == flag {
			fix = "NO_FIX"
			if uint32(flag)&2 != 0 {
				fix = "FIXED"
			}
		}
	}
	return fix, coordinate
}

type realtimeFrameReference struct {
	VIN, Protocol, EventID string
	ReceivedAt             time.Time
	Hydrogen, Location     bool
}

func realtimeReferenceKey(vin, protocol, eventID string) string {
	return vin + "\x00" + protocol + "\x00" + eventID
}

// An optional history outage must not make previously available realtime fields
// disappear. A shared deadline bounds enrichment even for the 2,000 VIN batch.
func (r *MySQLRepository) enrichRealtimeLiveData(ctx context.Context, vins []string, points map[string]RealtimeVehiclePoint, now time.Time) error {
	for vin, point := range points {
		point.LiveHydrogen = missingRealtimeHydrogen("")
		points[vin] = point
	}
	bounded, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()
	if err := r.loadRealtimeLiveData(bounded, vins, points, now); err != nil {
		log.Printf("openplatform realtime enrichment unavailable: type=%T reason=%s", err, realtimeEnrichmentErrorReason(err))
		// Keep successfully proven raw-frame values, and mark every unresolved
		// field missing/unknown rather than classifying a storage error unsupported.
		for vin, point := range points {
			if point.LiveHydrogen.HydrogenRecordTime == nil {
				point.LiveHydrogen = missingRealtimeHydrogen("")
			}
			points[vin] = point
		}
	}
	return nil
}

func (r *MySQLRepository) loadRealtimeLiveData(ctx context.Context, vins []string, points map[string]RealtimeVehiclePoint, now time.Time) error {
	for vin, point := range points {
		point.LiveHydrogen = missingRealtimeHydrogen(point.Protocol)
		points[vin] = point
	}
	// Repositories used without history retain compatible old fields and explicit
	// missing statuses; merged MySQL JSON is never used as a fallback.
	if r.tdengine == nil || r.tdDatabase == "" {
		return fmt.Errorf("realtime history is not configured")
	}
	refs := map[string]realtimeFrameReference{}
	for vin, point := range points {
		if point.LocationEventID != "" && !point.LocationReceivedAt.IsZero() {
			ref := realtimeFrameReference{VIN: vin, Protocol: point.Protocol, EventID: point.LocationEventID, ReceivedAt: point.LocationReceivedAt, Location: true}
			refs[realtimeReferenceKey(vin, ref.Protocol, ref.EventID)] = ref
		}
	}
	placeholders := strings.TrimRight(strings.Repeat("?,", len(vins)), ",")
	args := make([]any, len(vins))
	for i, vin := range vins {
		args[i] = vin
	}
	rows, err := r.db.QueryContext(ctx, `SELECT vin,protocol,event_id,received_at FROM vehicle_realtime_snapshot WHERE BINARY vin IN (`+placeholders+`) AND protocol='GB32960'`, args...)
	if err != nil {
		return err
	}
	for rows.Next() {
		var ref realtimeFrameReference
		var received sql.NullTime
		var eventID sql.NullString
		if err := rows.Scan(&ref.VIN, &ref.Protocol, &eventID, &received); err != nil {
			rows.Close()
			return err
		}
		point, exists := points[ref.VIN]
		if !exists {
			continue
		} // preserve existing realtime row selection contract
		point.LiveHydrogen = missingRealtimeHydrogen("GB32960")
		points[ref.VIN] = point
		if !received.Valid || !eventID.Valid || eventID.String == "" {
			continue
		}
		ref.EventID, ref.ReceivedAt, ref.Hydrogen = eventID.String, received.Time, true
		key := realtimeReferenceKey(ref.VIN, ref.Protocol, ref.EventID)
		if prior, ok := refs[key]; ok {
			ref.Location = prior.Location
		}
		refs[key] = ref
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return err
	}
	rows.Close()
	ordered := make([]realtimeFrameReference, 0, len(refs))
	for _, ref := range refs {
		ordered = append(ordered, ref)
	}

	// At most two references per requested VIN. Batches bound SQL size; primary
	// timestamp and VIN tag filters prevent unbounded history scans.
	frames, loadErr := r.loadRealtimeRawFrameBatches(ctx, ordered)
	for _, frame := range frames {
		ref, exists := refs[realtimeReferenceKey(frame.VIN, frame.Protocol, frame.EventID)]
		if !exists {
			continue
		}
		point, exists := points[frame.VIN]
		if !exists {
			continue
		}
		if ref.Hydrogen {
			var at time.Time
			if frame.EventMS.Valid && frame.EventMS.Int64 > 0 {
				at = time.UnixMilli(frame.EventMS.Int64)
			}
			point.LiveHydrogen = realtimeHydrogenFromFrame(frame.Parsed.String, at, now)
		}
		if ref.Location {
			point.GPSFixStatus, point.CoordinateSystem = realtimeGPSFromFrame(frame.Protocol, frame.Parsed.String)
		}
		points[frame.VIN] = point
	}
	if loadErr == nil && ctx.Err() == nil {
		var missing []realtimeFrameReference
		for _, ref := range refs {
			if ref.Hydrogen && points[ref.VIN].LiveHydrogen.HydrogenDataStatus == "MISSING" {
				missing = append(missing, ref)
			}
		}
		fallback, err := r.loadRealtimeRawFrameBatchesWithQuery(ctx, missing, realtimeHydrogenFallbackQuery, 1)
		applyRealtimeHydrogenFallback(points, missing, fallback, now)
		if err != nil {
			return err
		}
	}
	return loadErr
}

// Only explicitly missing hydrogen can be filled. In particular, the newest
// invalid measurement must never be hidden by an older valid reading.
func applyRealtimeHydrogenFallback(points map[string]RealtimeVehiclePoint, refs []realtimeFrameReference, frames []realtimeRawFrame, now time.Time) {
	allowed := make(map[string]bool, len(refs))
	for _, ref := range refs {
		allowed[ref.VIN] = true
	}
	// SQL orders by event_time, not arrival time: delayed retransmissions must not
	// displace newer measurements. Keep that rule explicit when processing rows.
	sort.SliceStable(frames, func(i, j int) bool { return frames[i].EventMS.Int64 > frames[j].EventMS.Int64 })
	for _, frame := range frames {
		if !allowed[frame.VIN] || frame.Protocol != "GB32960" {
			continue
		}
		point, exists := points[frame.VIN]
		if !exists || point.LiveHydrogen.HydrogenDataStatus != "MISSING" {
			continue
		}
		var at time.Time
		if frame.EventMS.Valid && frame.EventMS.Int64 > 0 {
			at = time.UnixMilli(frame.EventMS.Int64)
		}
		data := realtimeHydrogenFromFrame(frame.Parsed.String, at, now)
		if data.HydrogenDataStatus == "MISSING" {
			continue
		} // LIKE is a prefilter, never a JSON parser.
		point.LiveHydrogen = data
		points[frame.VIN] = point
	}
}

type realtimeRawFrame struct {
	VIN, Protocol, EventID string
	EventMS                sql.NullInt64
	Parsed                 sql.NullString
}

type realtimeRawBatchResult struct {
	frames []realtimeRawFrame
	err    error
}

// Four workers cap pressure on history storage while avoiding serial latency
// across a large authorized fleet. Only the caller writes the result map.
func (r *MySQLRepository) loadRealtimeRawFrameBatches(ctx context.Context, refs []realtimeFrameReference) ([]realtimeRawFrame, error) {
	return r.loadRealtimeRawFrameBatchesWithQuery(ctx, refs, realtimeRawFrameQuery, 100)
}

func (r *MySQLRepository) loadRealtimeRawFrameBatchesWithQuery(ctx context.Context, refs []realtimeFrameReference, buildQuery func(string, []realtimeFrameReference) (string, error), maxBatchSize int) ([]realtimeRawFrame, error) {
	batches := realtimeRawReferenceBatches(refs)
	if maxBatchSize == 1 {
		var singleVINBatches [][]realtimeFrameReference
		for _, batch := range batches {
			for i := range batch {
				singleVINBatches = append(singleVINBatches, batch[i:i+1])
			}
		}
		batches = singleVINBatches
	}
	batchCount := len(batches)
	if batchCount == 0 {
		return nil, nil
	}
	jobs := make(chan string, batchCount)
	results := make(chan realtimeRawBatchResult, batchCount)
	for _, batch := range batches {
		query, err := buildQuery(r.tdDatabase, batch)
		if err != nil {
			return nil, err
		}
		jobs <- query
	}
	close(jobs)
	workers := 4
	if batchCount < workers {
		workers = batchCount
	}
	for worker := 0; worker < workers; worker++ {
		go func() {
			for query := range jobs {
				if ctx.Err() != nil {
					return
				}
				frames, err := r.loadRealtimeRawFrameBatch(ctx, query)
				results <- realtimeRawBatchResult{frames: frames, err: err}
			}
		}()
	}
	var frames []realtimeRawFrame
	var firstErr error
	for batch := 0; batch < batchCount; batch++ {
		var result realtimeRawBatchResult
		select {
		case result = <-results:
		case <-ctx.Done():
			return frames, ctx.Err()
		}
		if firstErr == nil && result.err != nil {
			firstErr = result.err
		}
		frames = append(frames, result.frames...)
	}
	return frames, firstErr
}

func (r *MySQLRepository) loadRealtimeRawFrameBatch(ctx context.Context, query string) ([]realtimeRawFrame, error) {
	rows, err := r.tdengine.QueryContext(ctx, query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var frames []realtimeRawFrame
	for rows.Next() {
		var frame realtimeRawFrame
		if err := rows.Scan(&frame.VIN, &frame.Protocol, &frame.EventID, &frame.EventMS, &frame.Parsed); err != nil {
			return frames, err
		}
		frames = append(frames, frame)
	}
	return frames, rows.Err()
}

// Keep historical outliers away from current data: one query never spans more
// than five minutes of arrival time, even when a fleet includes months-old rows.
// Newest batches run first so the shared deadline favors currently active cars.
func realtimeRawReferenceBatches(refs []realtimeFrameReference) [][]realtimeFrameReference {
	ordered := append([]realtimeFrameReference(nil), refs...)
	sort.Slice(ordered, func(i, j int) bool { return ordered[i].ReceivedAt.After(ordered[j].ReceivedAt) })
	var batches [][]realtimeFrameReference
	for start := 0; start < len(ordered); {
		end := start + 1
		for end < len(ordered) && end-start < 100 && ordered[start].ReceivedAt.Sub(ordered[end].ReceivedAt) <= 5*time.Minute {
			end++
		}
		batches = append(batches, ordered[start:end])
		start = end
	}
	return batches
}

func realtimeRawFrameQuery(database string, refs []realtimeFrameReference) (string, error) {
	if !validTDIdentifier(database) || len(refs) == 0 || len(refs) > 100 {
		return "", fmt.Errorf("invalid realtime raw-frame query")
	}
	quote := func(s string) string { return "'" + strings.ReplaceAll(s, "'", "''") + "'" }
	clauses, vins, timestamps := make([]string, 0, len(refs)), make([]string, 0, len(refs)), make([]string, 0, len(refs))
	min, max := refs[0].ReceivedAt.UnixMilli(), refs[0].ReceivedAt.UnixMilli()
	for _, ref := range refs {
		at := ref.ReceivedAt.UnixMilli()
		if at < min {
			min = at
		}
		if at > max {
			max = at
		}
		vins = append(vins, quote(ref.VIN))
		timestamps = append(timestamps, strconv.FormatInt(at, 10))
		clauses = append(clauses, "(ts="+strconv.FormatInt(at, 10)+" AND vin="+quote(ref.VIN)+" AND protocol="+quote(ref.Protocol)+" AND event_id="+quote(ref.EventID)+")")
	}
	return `SELECT vin,protocol,event_id,CAST(event_time AS BIGINT),parsed_json FROM ` + database + `.raw_frames WHERE ts>=` + strconv.FormatInt(min, 10) + ` AND ts<=` + strconv.FormatInt(max, 10) + ` AND ts IN (` + strings.Join(timestamps, ",") + `) AND vin IN (` + strings.Join(vins, ",") + `) AND parse_status='OK' AND (` + strings.Join(clauses, " OR ") + `) LIMIT ` + strconv.Itoa(len(refs)*2), nil
}

// Search only five minutes before one snapshot arrival. TDengine applies LIMIT
// globally even with PARTITION BY, so each fallback query must contain one VIN.
// The shared worker pool still bounds concurrency to four and uses one deadline.
func realtimeHydrogenFallbackQuery(database string, refs []realtimeFrameReference) (string, error) {
	if !validTDIdentifier(database) || len(refs) != 1 {
		return "", fmt.Errorf("invalid realtime hydrogen fallback query")
	}
	quote := func(s string) string { return "'" + strings.ReplaceAll(s, "'", "''") + "'" }
	min, max := refs[0].ReceivedAt.Add(-5*time.Minute).UnixMilli(), refs[0].ReceivedAt.UnixMilli()
	var clauses, vins, fields []string
	for _, ref := range refs {
		start, end := ref.ReceivedAt.Add(-5*time.Minute).UnixMilli(), ref.ReceivedAt.UnixMilli()
		if start < min {
			min = start
		}
		if end > max {
			max = end
		}
		vins = append(vins, quote(ref.VIN))
		clauses = append(clauses, "(vin="+quote(ref.VIN)+" AND ts>="+strconv.FormatInt(start, 10)+" AND ts<="+strconv.FormatInt(end, 10)+")")
	}
	for _, key := range hydrogenMassFields {
		fields = append(fields, "parsed_json LIKE "+quote("%\""+key+"\":%"))
	}
	return `SELECT vin,protocol,event_id,CAST(event_time AS BIGINT),parsed_json FROM ` + database + `.raw_frames WHERE protocol='GB32960' AND ts>=` + strconv.FormatInt(min, 10) + ` AND ts<=` + strconv.FormatInt(max, 10) + ` AND vin IN (` + strings.Join(vins, ",") + `) AND parse_status='OK' AND (` + strings.Join(clauses, " OR ") + `) AND (` + strings.Join(fields, " OR ") + `) ORDER BY event_time DESC,ts DESC LIMIT 1`, nil
}

var realtimeErrorQuotedText = regexp.MustCompile(`'[^']*'|"[^"]*"`)
var realtimeErrorVIN = regexp.MustCompile(`\b[A-Z0-9]{12,32}\b`)

// Driver errors sometimes append the whole query. Keep the failure reason for
// operations without writing fleet identifiers or SQL literals to the log.
func realtimeEnrichmentErrorReason(err error) string {
	reason := err.Error()
	upper := strings.ToUpper(reason)
	for _, keyword := range []string{"SELECT ", "INSERT ", "UPDATE ", "DELETE "} {
		if index := strings.Index(upper, keyword); index >= 0 {
			reason = reason[:index] + "[SQL omitted]"
			break
		}
	}
	reason = realtimeErrorQuotedText.ReplaceAllString(reason, "[quoted value omitted]")
	reason = realtimeErrorVIN.ReplaceAllString(reason, "[VIN omitted]")
	reason = strings.Join(strings.Fields(reason), " ")
	if len(reason) > 300 {
		reason = reason[:300] + "..."
	}
	return reason
}