lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime/repository.go

package realtime

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/redis/go-redis/v9"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)

type Repository struct {
	client *redis.Client
	cfg    Config
}

func NewRepository(client *redis.Client, cfg Config) *Repository {
	if client == nil {
		panic("redis client must not be nil")
	}
	return &Repository{client: client, cfg: cfg}
}

func (r *Repository) FastUpdate(ctx context.Context, env envelope.FrameEnvelope) error {
	vin := strings.TrimSpace(env.VIN)
	if vin == "" {
		return nil
	}
	vehicleKey := strings.TrimSpace(env.VehicleKey())
	if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") {
		return nil
	}
	if err := r.setKV(ctx, vin, env, env.Parsed); err != nil {
		return err
	}
	return r.setOnline(ctx, vehicleKey, vin, []envelope.Protocol{env.Protocol}, env.ReceivedAtMS)
}

func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) error {
	vin := strings.TrimSpace(env.VIN)
	if vin == "" {
		return nil
	}
	vehicleKey := strings.TrimSpace(env.VehicleKey())
	if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") {
		return nil
	}
	nowMS := time.Now().UnixMilli()
	eventMS := env.EventTimeMS
	if eventMS <= 0 {
		eventMS = env.ReceivedAtMS
	}
	protocolSnapshot := Snapshot{
		VehicleKey:   vehicleKey,
		VIN:          vin,
		Protocol:     env.Protocol,
		EventID:      env.StableEventID(),
		EventTimeMS:  eventMS,
		ReceivedAtMS: env.ReceivedAtMS,
		Fields:       cloneFields(env.Fields),
		FieldTimesMS: fieldTimes(env.Fields, eventMS),
		Parsed:       cloneMap(env.Parsed),
		UpdatedAtMS:  nowMS,
	}
	existingProtocol, err := r.GetProtocol(ctx, vehicleKey, env.Protocol)
	if err != nil && !errors.Is(err, redis.Nil) {
		return err
	}
	existingParsed, err := r.GetRealtimeRaw(ctx, vehicleKey, env.Protocol)
	if err != nil && !errors.Is(err, redis.Nil) {
		return err
	}
	if existingProtocol.VehicleKey != "" {
		protocolSnapshot = existingProtocol
		if protocolSnapshot.VIN == "" && vin != "" {
			protocolSnapshot.VIN = vin
		}
		mergeFields(&protocolSnapshot, env.Fields, eventMS)
		if eventMS >= protocolSnapshot.EventTimeMS {
			protocolSnapshot.EventTimeMS = eventMS
			protocolSnapshot.EventID = env.StableEventID()
			protocolSnapshot.ReceivedAtMS = env.ReceivedAtMS
		}
		protocolSnapshot.UpdatedAtMS = nowMS
	}
	protocolSnapshot.Parsed = mergeParsedForProtocol(env.Protocol, existingParsed, env.Parsed)
	if err := r.setJSON(ctx, protocolKey(vehicleKey, env.Protocol), protocolSnapshot.Lightweight(), r.cfg.ttl()); err != nil {
		return err
	}
	if err := r.setJSON(ctx, realtimeRawKey(vehicleKey, env.Protocol), protocolSnapshot.Parsed, r.cfg.ttl()); err != nil {
		return err
	}
	if err := r.setKV(ctx, vin, env, protocolSnapshot.Parsed); err != nil {
		return err
	}

	protocols, err := r.addProtocol(ctx, vehicleKey, env.Protocol)
	if err != nil {
		return err
	}

	merged, err := r.GetMerged(ctx, vehicleKey)
	if err != nil && !errors.Is(err, redis.Nil) {
		return err
	}
	if merged.VehicleKey == "" {
		merged = Snapshot{
			VehicleKey:   vehicleKey,
			VIN:          vin,
			Fields:       map[string]any{},
			FieldTimesMS: map[string]int64{},
		}
	} else if merged.VIN == "" && vin != "" {
		merged.VIN = vin
	}
	mergeFields(&merged, env.Fields, eventMS)
	if eventMS >= merged.EventTimeMS {
		merged.EventTimeMS = eventMS
		merged.EventID = env.StableEventID()
		merged.ReceivedAtMS = env.ReceivedAtMS
	}
	merged.Protocols = protocols
	merged.UpdatedAtMS = nowMS
	if err := r.setJSON(ctx, mergedKey(vehicleKey), merged, r.cfg.ttl()); err != nil {
		return err
	}

	online := OnlineStatus{
		VehicleKey: vehicleKey,
		VIN:        vin,
		Online:     true,
		LastSeenMS: env.ReceivedAtMS,
		Protocols:  protocols,
		TTLSeconds: int64(r.cfg.ttl().Seconds()),
	}
	if err := r.setOnlineStatus(ctx, online); err != nil {
		return err
	}
	return r.client.ZAdd(ctx, "vehicle:last_seen", redis.Z{Score: float64(env.ReceivedAtMS), Member: vehicleKey}).Err()
}

func (r *Repository) GetMerged(ctx context.Context, vehicleKey string) (Snapshot, error) {
	return r.getSnapshot(ctx, mergedKey(vehicleKey))
}

func (r *Repository) GetProtocol(ctx context.Context, vehicleKey string, protocol envelope.Protocol) (Snapshot, error) {
	return r.getSnapshot(ctx, protocolKey(vehicleKey, protocol))
}

func (r *Repository) GetRealtimeRaw(ctx context.Context, vehicleKey string, protocol envelope.Protocol) (map[string]any, error) {
	payload, err := r.client.Get(ctx, realtimeRawKey(vehicleKey, protocol)).Bytes()
	if err != nil {
		return nil, err
	}
	var out map[string]any
	return out, json.Unmarshal(payload, &out)
}

func (r *Repository) IsOnline(ctx context.Context, vehicleKey string) (OnlineStatus, error) {
	vehicleKey = strings.TrimSpace(vehicleKey)
	var status OnlineStatus
	payload, err := r.client.Get(ctx, onlineKey(vehicleKey)).Bytes()
	if err != nil {
		if errors.Is(err, redis.Nil) {
			return OnlineStatus{VehicleKey: vehicleKey, VIN: vehicleKey, Online: false}, nil
		}
		return status, err
	}
	if err := json.Unmarshal(payload, &status); err != nil {
		return status, err
	}
	ttl := r.client.TTL(ctx, onlineKey(vehicleKey)).Val()
	status.Online = ttl > 0
	status.TTLSeconds = int64(ttl.Seconds())
	return status, nil
}

func (r *Repository) getSnapshot(ctx context.Context, key string) (Snapshot, error) {
	var snapshot Snapshot
	payload, err := r.client.Get(ctx, key).Bytes()
	if err != nil {
		return snapshot, err
	}
	return snapshot, json.Unmarshal(payload, &snapshot)
}

func (r *Repository) setJSON(ctx context.Context, key string, value any, ttl time.Duration) error {
	payload, err := json.Marshal(value)
	if err != nil {
		return err
	}
	return r.client.Set(ctx, key, payload, ttl).Err()
}

func (r *Repository) setKV(ctx context.Context, vin string, env envelope.FrameEnvelope, parsed map[string]any) error {
	rows := realtimeKVFields(env, parsed)
	if len(rows) == 0 {
		return nil
	}
	grouped := make(map[string]map[string]any)
	for _, row := range rows {
		key := realtimeKVKey(row.Protocol, vin, row.Domain)
		values := grouped[key]
		if values == nil {
			values = map[string]any{
				"_event_time_ms":  strconv.FormatInt(row.EventTimeMS, 10),
				"_received_at_ms": strconv.FormatInt(row.ReceivedAtMS, 10),
				"_event_id":       row.EventID,
				"_protocol":       string(row.Protocol),
				"_vin":            vin,
				"_domain":         row.Domain,
			}
			grouped[key] = values
		}
		values[row.Field] = row.Value
		values["_value_type:"+row.Field] = row.ValueType
	}
	pipe := r.client.Pipeline()
	for key, values := range grouped {
		pipe.HSet(ctx, key, values)
	}
	_, err := pipe.Exec(ctx)
	return err
}

func (r *Repository) setOnline(ctx context.Context, vehicleKey string, vin string, protocols []envelope.Protocol, lastSeenMS int64) error {
	online := OnlineStatus{
		VehicleKey: vehicleKey,
		VIN:        vin,
		Online:     true,
		LastSeenMS: lastSeenMS,
		Protocols:  protocols,
		TTLSeconds: int64(r.cfg.ttl().Seconds()),
	}
	if err := r.setOnlineStatus(ctx, online); err != nil {
		return err
	}
	return r.client.ZAdd(ctx, "vehicle:last_seen", redis.Z{Score: float64(lastSeenMS), Member: vehicleKey}).Err()
}

func (r *Repository) setOnlineStatus(ctx context.Context, online OnlineStatus) error {
	return r.setJSON(ctx, onlineKey(online.VehicleKey), online, r.cfg.ttl())
}

func (r *Repository) addProtocol(ctx context.Context, vehicleKey string, protocol envelope.Protocol) ([]envelope.Protocol, error) {
	key := protocolsKey(vehicleKey)
	if err := r.client.SAdd(ctx, key, string(protocol)).Err(); err != nil {
		return nil, err
	}
	_ = r.client.Expire(ctx, key, r.cfg.ttl()).Err()
	values, err := r.client.SMembers(ctx, key).Result()
	if err != nil {
		return nil, err
	}
	sort.Strings(values)
	out := make([]envelope.Protocol, 0, len(values))
	for _, value := range values {
		out = append(out, envelope.Protocol(value))
	}
	return out, nil
}

func mergeFields(snapshot *Snapshot, fields map[string]any, eventMS int64) {
	if snapshot.Fields == nil {
		snapshot.Fields = map[string]any{}
	}
	if snapshot.FieldTimesMS == nil {
		snapshot.FieldTimesMS = map[string]int64{}
	}
	for key, value := range fields {
		if key == envelope.FieldTotalMileageKM && !positiveNumber(value) {
			continue
		}
		if eventMS >= snapshot.FieldTimesMS[key] {
			snapshot.Fields[key] = value
			snapshot.FieldTimesMS[key] = eventMS
		}
	}
}

func positiveNumber(value any) bool {
	switch typed := value.(type) {
	case float64:
		return typed > 0
	case float32:
		return typed > 0
	case int:
		return typed > 0
	case int64:
		return typed > 0
	case uint16:
		return typed > 0
	case uint32:
		return typed > 0
	case string:
		parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64)
		return err == nil && parsed > 0
	default:
		return false
	}
}

func cloneFields(fields map[string]any) map[string]any {
	if len(fields) == 0 {
		return map[string]any{}
	}
	out := make(map[string]any, len(fields))
	for key, value := range fields {
		out[key] = value
	}
	return out
}

func cloneMap(values map[string]any) map[string]any {
	if len(values) == 0 {
		return map[string]any{}
	}
	out := make(map[string]any, len(values))
	for key, value := range values {
		out[key] = cloneAny(value)
	}
	return out
}

func cloneAny(value any) any {
	switch typed := value.(type) {
	case map[string]any:
		return cloneMap(typed)
	case []any:
		out := make([]any, len(typed))
		for i, item := range typed {
			out[i] = cloneAny(item)
		}
		return out
	case []map[string]any:
		out := make([]any, len(typed))
		for i, item := range typed {
			out[i] = cloneMap(item)
		}
		return out
	default:
		return value
	}
}

func mergeParsed(existing map[string]any, incoming map[string]any) map[string]any {
	out := cloneMap(existing)
	for key, value := range incoming {
		out[key] = mergeAny(out[key], value)
	}
	return out
}

func mergeParsedForProtocol(protocol envelope.Protocol, existing map[string]any, incoming map[string]any) map[string]any {
	out := mergeParsed(existing, incoming)
	for _, key := range replaceParsedKeys(protocol, incoming) {
		out[key] = cloneAny(incoming[key])
	}
	if protocol == envelope.ProtocolGB32960 {
		normalizeGB32960RealtimeParsed(out)
	}
	return out
}

func replaceParsedKeys(protocol envelope.Protocol, incoming map[string]any) []string {
	keys := make([]string, 0, 2)
	if _, ok := incoming["header"]; ok {
		keys = append(keys, "header")
	}
	if protocol == envelope.ProtocolJT808 {
		if _, ok := incoming["location"]; ok {
			keys = append(keys, "location")
		}
	}
	return keys
}

func mergeAny(existing any, incoming any) any {
	switch incomingTyped := incoming.(type) {
	case map[string]any:
		existingMap, _ := existing.(map[string]any)
		return mergeParsed(existingMap, incomingTyped)
	case []any:
		if merged, ok := mergeMapSlice(existing, incomingTyped); ok {
			return merged
		}
		return cloneAny(incomingTyped)
	case []map[string]any:
		items := make([]any, len(incomingTyped))
		for i, item := range incomingTyped {
			items[i] = item
		}
		if merged, ok := mergeMapSlice(existing, items); ok {
			return merged
		}
		return cloneAny(incomingTyped)
	default:
		return cloneAny(incoming)
	}
}

func mergeMapSlice(existing any, incoming []any) ([]any, bool) {
	existingItems, ok := asAnySlice(existing)
	if !ok {
		return nil, false
	}
	merged := make([]any, 0, len(existingItems)+len(incoming))
	index := map[string]int{}
	add := func(item any) {
		itemMap, ok := item.(map[string]any)
		if !ok {
			merged = append(merged, cloneAny(item))
			return
		}
		key := mergeSliceKey(itemMap)
		if key == "" {
			merged = append(merged, cloneMap(itemMap))
			return
		}
		if pos, exists := index[key]; exists {
			merged[pos] = mergeAny(merged[pos], itemMap)
			return
		}
		index[key] = len(merged)
		merged = append(merged, cloneMap(itemMap))
	}
	for _, item := range existingItems {
		add(item)
	}
	for _, item := range incoming {
		add(item)
	}
	return merged, true
}

func asAnySlice(value any) ([]any, bool) {
	switch typed := value.(type) {
	case []any:
		return typed, true
	case []map[string]any:
		out := make([]any, len(typed))
		for i, item := range typed {
			out[i] = item
		}
		return out, true
	default:
		return nil, false
	}
}

func mergeSliceKey(item map[string]any) string {
	for _, key := range []string{"name", "type", "id", "serial_no"} {
		if value, ok := item[key]; ok {
			text := strings.TrimSpace(strconvAny(value))
			if text != "" {
				return key + ":" + text
			}
		}
	}
	return ""
}

func strconvAny(value any) string {
	switch typed := value.(type) {
	case string:
		return typed
	default:
		return fmt.Sprint(typed)
	}
}

func normalizeGB32960RealtimeParsed(parsed map[string]any) {
	units, ok := asAnySlice(parsed["data_units"])
	if !ok {
		return
	}
	for _, unit := range units {
		unitMap, ok := unit.(map[string]any)
		if !ok {
			continue
		}
		value, ok := unitMap["value"].(map[string]any)
		if !ok {
			continue
		}
		switch strings.TrimSpace(strconvAny(unitMap["name"])) {
		case "gd_fc_stack":
			collapseGDFCStackValue(value)
		case "gd_fc_auxiliary":
			collapseGDFCAuxiliaryValue(value)
		}
	}
}

func collapseGDFCStackValue(value map[string]any) {
	summaries, ok := asAnySlice(value["summaries"])
	if !ok || len(summaries) == 0 {
		return
	}
	collapsed := map[string]any{}
	for _, item := range summaries {
		summary, ok := item.(map[string]any)
		if !ok {
			continue
		}
		for key, fieldValue := range summary {
			if isGDFCStackFragmentField(key) {
				continue
			}
			collapsed[key] = cloneAny(fieldValue)
		}
	}
	if len(collapsed) == 0 {
		delete(value, "summaries")
		return
	}
	value["stack_count"] = 1
	value["summaries"] = []any{collapsed}
}

func isGDFCStackFragmentField(key string) bool {
	switch key {
	case "frame_cell_start", "frame_cell_count", "frame_max_cell_voltage_v", "frame_min_cell_voltage_v":
		return true
	default:
		return false
	}
}

func collapseGDFCAuxiliaryValue(value map[string]any) {
	subsystems, ok := asAnySlice(value["subsystems"])
	if !ok || len(subsystems) == 0 {
		return
	}
	latest := subsystems[len(subsystems)-1]
	subsystem, ok := latest.(map[string]any)
	if !ok {
		return
	}
	value["subsystem_count"] = 1
	value["subsystems"] = []any{cloneMap(subsystem)}
}

func fieldTimes(fields map[string]any, eventMS int64) map[string]int64 {
	out := make(map[string]int64, len(fields))
	for key := range fields {
		out[key] = eventMS
	}
	return out
}

func mergedKey(vehicleKey string) string {
	return "vehicle:latest:" + strings.TrimSpace(vehicleKey)
}

func protocolKey(vehicleKey string, protocol envelope.Protocol) string {
	return "vehicle:latest:" + strings.TrimSpace(vehicleKey) + ":" + string(protocol)
}

func realtimeRawKey(vehicleKey string, protocol envelope.Protocol) string {
	return "vehicle:realtime-raw:" + string(protocol) + ":" + strings.TrimSpace(vehicleKey)
}

func realtimeKVKey(protocol envelope.Protocol, vin string, domain string) string {
	return "vehicle:rt-kv:" + string(protocol) + ":" + strings.TrimSpace(vin) + ":" + strings.TrimSpace(domain)
}

func onlineKey(vehicleKey string) string {
	return "vehicle:online:" + strings.TrimSpace(vehicleKey)
}

func protocolsKey(vehicleKey string) string {
	return "vehicle:protocols:" + strings.TrimSpace(vehicleKey)
}