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
}

type FastUpdateResult struct {
	EnvelopesSeen                     int
	EnvelopesUpdated                  int
	EnvelopesSkippedNonRealtime       int
	EnvelopesSkippedMissingVIN        int
	EnvelopesSkippedMissingVehicleKey int
	EnvelopesSkippedMissingFields     int
	FieldsSeen                        int
	FieldsWritten                     int
	FieldsSkippedStale                int
}

func (r FastUpdateResult) Add(other FastUpdateResult) FastUpdateResult {
	return FastUpdateResult{
		EnvelopesSeen:                     r.EnvelopesSeen + other.EnvelopesSeen,
		EnvelopesUpdated:                  r.EnvelopesUpdated + other.EnvelopesUpdated,
		EnvelopesSkippedNonRealtime:       r.EnvelopesSkippedNonRealtime + other.EnvelopesSkippedNonRealtime,
		EnvelopesSkippedMissingVIN:        r.EnvelopesSkippedMissingVIN + other.EnvelopesSkippedMissingVIN,
		EnvelopesSkippedMissingVehicleKey: r.EnvelopesSkippedMissingVehicleKey + other.EnvelopesSkippedMissingVehicleKey,
		EnvelopesSkippedMissingFields:     r.EnvelopesSkippedMissingFields + other.EnvelopesSkippedMissingFields,
		FieldsSeen:                        r.FieldsSeen + other.FieldsSeen,
		FieldsWritten:                     r.FieldsWritten + other.FieldsWritten,
		FieldsSkippedStale:                r.FieldsSkippedStale + other.FieldsSkippedStale,
	}
}

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 {
	_, err := r.FastUpdateWithResult(ctx, env)
	return err
}

func (r *Repository) FastUpdateWithResult(ctx context.Context, env envelope.FrameEnvelope) (FastUpdateResult, error) {
	result := FastUpdateResult{EnvelopesSeen: 1}
	if !envelope.IsRealtimeTelemetryFrame(env) {
		result.EnvelopesSkippedNonRealtime = 1
		return result, nil
	}
	vin := strings.TrimSpace(env.VIN)
	if vin == "" {
		result.EnvelopesSkippedMissingVIN = 1
		return result, nil
	}
	vehicleKey := strings.TrimSpace(env.VehicleKey())
	if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") {
		result.EnvelopesSkippedMissingVehicleKey = 1
		return result, nil
	}
	if len(env.ParsedFields) == 0 {
		result.EnvelopesSkippedMissingFields = 1
		return result, nil
	}
	return r.setFastProjection(ctx, vehicleKey, vin, env)
}

func (r *Repository) FastUpdateBatch(ctx context.Context, envs []envelope.FrameEnvelope) error {
	_, err := r.FastUpdateBatchWithResult(ctx, envs)
	return err
}

func (r *Repository) FastUpdateBatchWithResult(ctx context.Context, envs []envelope.FrameEnvelope) (FastUpdateResult, error) {
	if len(envs) == 0 {
		return FastUpdateResult{}, nil
	}
	pipe := r.client.Pipeline()
	var result FastUpdateResult
	queued := make([]queuedFastProjection, 0, len(envs))
	for _, env := range envs {
		result.EnvelopesSeen++
		if !envelope.IsRealtimeTelemetryFrame(env) {
			result.EnvelopesSkippedNonRealtime++
			continue
		}
		vin := strings.TrimSpace(env.VIN)
		if vin == "" {
			result.EnvelopesSkippedMissingVIN++
			continue
		}
		vehicleKey := strings.TrimSpace(env.VehicleKey())
		if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") {
			result.EnvelopesSkippedMissingVehicleKey++
			continue
		}
		if len(env.ParsedFields) == 0 {
			result.EnvelopesSkippedMissingFields++
			continue
		}
		queuedProjection, err := r.queueFastProjection(ctx, pipe, vehicleKey, vin, env)
		if err != nil {
			return FastUpdateResult{}, err
		}
		queued = append(queued, queuedProjection)
	}
	if len(queued) == 0 {
		return result, nil
	}
	_, err := pipe.Exec(ctx)
	if err != nil {
		return FastUpdateResult{}, err
	}
	for _, item := range queued {
		result = result.Add(item.result())
	}
	return result, nil
}

func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) error {
	if !envelope.IsRealtimeTelemetryFrame(env) {
		return nil
	}
	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, _ := envelope.NormalizedEventTimeMS(env)
	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); 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,
		Protocol:   env.Protocol,
		Online:     true,
		LastSeenMS: env.ReceivedAtMS,
		Protocols:  protocols,
		TTLSeconds: int64(r.cfg.ttl().Seconds()),
	}
	if err := r.setOnlineStatus(ctx, online); err != nil {
		return err
	}
	return nil
}

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)
	best := OnlineStatus{VehicleKey: vehicleKey, VIN: vehicleKey, Online: false}
	for _, protocol := range knownRealtimeProtocols() {
		key := onlineKey(protocol, vehicleKey)
		payload, err := r.client.Get(ctx, key).Bytes()
		if err != nil {
			if errors.Is(err, redis.Nil) {
				continue
			}
			return OnlineStatus{}, err
		}
		var status OnlineStatus
		if err := json.Unmarshal(payload, &status); err != nil {
			return OnlineStatus{}, err
		}
		ttl := r.client.TTL(ctx, key).Val()
		status.Online = ttl > 0
		status.TTLSeconds = int64(ttl.Seconds())
		if status.Protocol == "" {
			status.Protocol = protocol
		}
		if status.LastSeenMS >= best.LastSeenMS {
			best = status
		}
	}
	return best, nil
}

func (r *Repository) ListOnline(ctx context.Context, query OnlineListQuery) ([]OnlineStatus, int64, error) {
	query = normalizeOnlineListQuery(query)
	total, err := r.onlineTotal(ctx, query.Protocol)
	if err != nil {
		return nil, 0, err
	}
	items := make([]OnlineStatus, 0, query.Limit)
	start := int64(query.Offset)
	if query.Protocol != "" {
		start = 0
	}
	skipped := 0
	batchSize := int64(query.Limit * 5)
	if batchSize < 100 {
		batchSize = 100
	}
	for len(items) < query.Limit {
		stop := start + batchSize - 1
		members, err := r.client.ZRevRangeWithScores(ctx, "vehicle:last_seen", start, stop).Result()
		if err != nil {
			return nil, 0, err
		}
		if len(members) == 0 {
			break
		}
		for _, member := range members {
			protocol, vin, ok := splitOnlineMember(fmt.Sprint(member.Member))
			if !ok {
				continue
			}
			if query.Protocol != "" && protocol != query.Protocol {
				continue
			}
			if query.Protocol != "" && skipped < query.Offset {
				skipped++
				continue
			}
			status, err := r.onlineStatusFor(ctx, protocol, vin, int64(member.Score))
			if err != nil {
				return nil, 0, err
			}
			items = append(items, status)
			if len(items) >= query.Limit {
				break
			}
		}
		if int64(len(members)) < batchSize {
			break
		}
		start += batchSize
	}
	return items, total, nil
}

func (r *Repository) PipelineSummary(ctx context.Context) (PipelineSummary, error) {
	out := PipelineSummary{
		Protocols: make([]ProtocolPipelineSummary, 0, len(knownRealtimeProtocols())),
		UpdatedMS: time.Now().UnixMilli(),
	}
	for _, protocol := range knownRealtimeProtocols() {
		indexed, err := r.client.SCard(ctx, realtimeIndexKey(protocol)).Result()
		if err != nil {
			return PipelineSummary{}, err
		}
		onlineCount, latestSeen, err := r.protocolOnlineStats(ctx, protocol)
		if err != nil {
			return PipelineSummary{}, err
		}
		out.Protocols = append(out.Protocols, ProtocolPipelineSummary{
			Protocol:     protocol,
			IndexedCount: indexed,
			OnlineCount:  onlineCount,
			LatestSeenMS: latestSeen,
		})
	}
	return out, nil
}

func (r *Repository) onlineTotal(ctx context.Context, protocol envelope.Protocol) (int64, error) {
	if protocol != "" {
		return r.client.SCard(ctx, realtimeIndexKey(protocol)).Result()
	}
	return r.client.ZCard(ctx, "vehicle:last_seen").Result()
}

func (r *Repository) protocolOnlineStats(ctx context.Context, protocol envelope.Protocol) (int64, int64, error) {
	members, err := r.client.SMembers(ctx, realtimeIndexKey(protocol)).Result()
	if err != nil {
		return 0, 0, err
	}
	var onlineCount int64
	var latestSeen int64
	for _, vin := range members {
		status, err := r.onlineStatusFor(ctx, protocol, vin, 0)
		if err != nil {
			return 0, 0, err
		}
		if status.Online {
			onlineCount++
		}
		if status.LastSeenMS > latestSeen {
			latestSeen = status.LastSeenMS
		}
	}
	return onlineCount, latestSeen, nil
}

func (r *Repository) onlineStatusFor(ctx context.Context, protocol envelope.Protocol, vin string, scoreMS int64) (OnlineStatus, error) {
	key := onlineKey(protocol, vin)
	stateKey := onlineStateKey(protocol, vin)
	state, err := r.client.HGetAll(ctx, stateKey).Result()
	if err != nil {
		return OnlineStatus{}, err
	}
	ttl := r.client.TTL(ctx, key).Val()
	status := OnlineStatus{
		VehicleKey:     firstNonEmptyString(state["vehicle_key"], vin),
		VIN:            firstNonEmptyString(state["vin"], vin),
		Protocol:       protocol,
		Online:         ttl > 0,
		LastSeenMS:     firstPositiveInt64(parseInt64(state["last_seen_ms"]), scoreMS),
		OfflineAfterMS: parseInt64(state["offline_after_ms"]),
		SourceEndpoint: state["source_endpoint"],
		Protocols:      []envelope.Protocol{protocol},
		TTLSeconds:     int64(ttl.Seconds()),
	}
	if status.TTLSeconds < 0 {
		status.TTLSeconds = 0
	}
	if status.OfflineAfterMS <= 0 && status.LastSeenMS > 0 {
		status.OfflineAfterMS = status.LastSeenMS + r.cfg.ttl().Milliseconds()
	}
	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) error {
	values, types := realtimeKVMapsForEnvelope(env)
	if len(values) == 0 {
		return nil
	}
	meta := map[string]any{
		"event_time_ms":   strconv.FormatInt(eventTimeOrReceivedMS(env), 10),
		"received_at_ms":  strconv.FormatInt(env.ReceivedAtMS, 10),
		"event_id":        env.StableEventID(),
		"protocol":        string(env.Protocol),
		"vin":             vin,
		"source_endpoint": env.SourceEndpoint,
		"field_mapping":   realtimeFieldMappingVersion,
	}
	return evalGuardedRealtimeKV(ctx, r.client, env.Protocol, vin, eventTimeOrReceivedMS(env), values, types, meta).Err()
}

func (r *Repository) setFastProjection(ctx context.Context, vehicleKey string, vin string, env envelope.FrameEnvelope) (FastUpdateResult, error) {
	pipe := r.client.Pipeline()
	queued, err := r.queueFastProjection(ctx, pipe, vehicleKey, vin, env)
	if err != nil {
		return FastUpdateResult{}, err
	}
	if _, err := pipe.Exec(ctx); err != nil {
		return FastUpdateResult{}, err
	}
	result := queued.result()
	result.EnvelopesSeen = 1
	return result, nil
}

type queuedFastProjection struct {
	fieldsSeen int
	writeCmd   *redis.Cmd
}

func (q queuedFastProjection) result() FastUpdateResult {
	written := redisCmdInt(q.writeCmd)
	skipped := q.fieldsSeen - written
	if skipped < 0 {
		skipped = 0
	}
	return FastUpdateResult{
		EnvelopesUpdated:   1,
		FieldsSeen:         q.fieldsSeen,
		FieldsWritten:      written,
		FieldsSkippedStale: skipped,
	}
}

func (r *Repository) queueFastProjection(ctx context.Context, pipe redis.Pipeliner, vehicleKey string, vin string, env envelope.FrameEnvelope) (queuedFastProjection, error) {
	values, types := realtimeKVMapsForEnvelope(env)
	eventTimeMS := eventTimeOrReceivedMS(env)
	offlineAfterMS := env.ReceivedAtMS + r.cfg.ttl().Milliseconds()
	online := OnlineStatus{
		VehicleKey:     vehicleKey,
		VIN:            vin,
		Protocol:       env.Protocol,
		Online:         true,
		LastSeenMS:     env.ReceivedAtMS,
		OfflineAfterMS: offlineAfterMS,
		SourceEndpoint: env.SourceEndpoint,
		Protocols:      []envelope.Protocol{env.Protocol},
		TTLSeconds:     int64(r.cfg.ttl().Seconds()),
	}
	payload, err := json.Marshal(online)
	if err != nil {
		return queuedFastProjection{}, err
	}
	meta := map[string]any{
		"event_time_ms":   strconv.FormatInt(eventTimeMS, 10),
		"received_at_ms":  strconv.FormatInt(env.ReceivedAtMS, 10),
		"event_id":        env.StableEventID(),
		"protocol":        string(env.Protocol),
		"vin":             vin,
		"source_endpoint": env.SourceEndpoint,
		"field_mapping":   realtimeFieldMappingVersion,
	}
	state := map[string]any{
		"vehicle_key":      vehicleKey,
		"vin":              vin,
		"protocol":         string(env.Protocol),
		"online":           strconv.FormatBool(true),
		"last_seen_ms":     strconv.FormatInt(env.ReceivedAtMS, 10),
		"offline_after_ms": strconv.FormatInt(offlineAfterMS, 10),
		"ttl_seconds":      strconv.FormatInt(int64(r.cfg.ttl().Seconds()), 10),
		"source_endpoint":  env.SourceEndpoint,
	}
	queued := queuedFastProjection{fieldsSeen: len(values)}
	if len(values) > 0 {
		queued.writeCmd = evalGuardedRealtimeKV(ctx, pipe, env.Protocol, vin, eventTimeMS, values, types, meta)
	}
	if _, err := evalGuardedOnlineStatus(ctx, pipe, online, state, payload, r.cfg.ttl()); err != nil {
		return queuedFastProjection{}, err
	}
	return queued, nil
}

const guardedRealtimeKVScript = `
local incoming = tonumber(ARGV[1]) or 0
local value_count = tonumber(ARGV[2]) or 0
local idx = 3
local written = 0

for i = 1, value_count do
  local field = ARGV[idx]
  local value = ARGV[idx + 1]
  idx = idx + 2
  local current = tonumber(redis.call('HGET', KEYS[3], field) or '') or 0
  if current <= incoming then
    redis.call('HSET', KEYS[1], field, value)
    redis.call('HSET', KEYS[3], field, incoming)
    written = written + 1
  end
end

local type_count = tonumber(ARGV[idx]) or 0
idx = idx + 1
for i = 1, type_count do
  local field = ARGV[idx]
  local value_type = ARGV[idx + 1]
  idx = idx + 2
  local current = tonumber(redis.call('HGET', KEYS[3], field) or '') or 0
  if current <= incoming then
    redis.call('HSET', KEYS[2], field, value_type)
  end
end

local meta_count = tonumber(ARGV[idx]) or 0
idx = idx + 1
local current_meta = tonumber(redis.call('HGET', KEYS[4], 'event_time_ms') or '') or 0
if current_meta <= incoming then
  for i = 1, meta_count do
    redis.call('HSET', KEYS[4], ARGV[idx], ARGV[idx + 1])
    idx = idx + 2
  end
end

return written
`

const guardedOnlineStatusScript = `
local incoming = tonumber(ARGV[1]) or 0
local ttl_ms = tonumber(ARGV[2]) or 60000
local payload = ARGV[3]
local member = ARGV[4]
local vin = ARGV[5]
local state_count = tonumber(ARGV[6]) or 0
local idx = 7
local current = tonumber(redis.call('HGET', KEYS[2], 'last_seen_ms') or '') or 0

redis.call('SADD', KEYS[4], vin)
if current <= incoming then
  redis.call('SET', KEYS[1], payload, 'PX', ttl_ms)
  for i = 1, state_count do
    redis.call('HSET', KEYS[2], ARGV[idx], ARGV[idx + 1])
    idx = idx + 2
  end
  redis.call('ZADD', KEYS[3], incoming, member)
  return 1
end
return 0
`

type redisEvaler interface {
	Eval(ctx context.Context, script string, keys []string, args ...any) *redis.Cmd
}

func redisCmdInt(cmd *redis.Cmd) int {
	if cmd == nil {
		return 0
	}
	value, err := cmd.Int64()
	if err != nil {
		return 0
	}
	return int(value)
}

func evalGuardedRealtimeKV(ctx context.Context, evaler redisEvaler, protocol envelope.Protocol, vin string, eventTimeMS int64, values map[string]any, types map[string]any, meta map[string]any) *redis.Cmd {
	keys := []string{
		realtimeKVValuesKey(protocol, vin),
		realtimeKVTypesKey(protocol, vin),
		realtimeKVTimesKey(protocol, vin),
		realtimeKVMetaKey(protocol, vin),
	}
	args := guardedRealtimeKVArgs(eventTimeMS, values, types, meta)
	return evaler.Eval(ctx, guardedRealtimeKVScript, keys, args...)
}

func evalGuardedOnlineStatus(ctx context.Context, evaler redisEvaler, online OnlineStatus, state map[string]any, payload []byte, ttl time.Duration) (*redis.Cmd, error) {
	protocol := online.Protocol
	if protocol == "" && len(online.Protocols) > 0 {
		protocol = online.Protocols[0]
	}
	vin := strings.TrimSpace(online.VIN)
	if protocol == "" || vin == "" {
		return nil, nil
	}
	if len(payload) == 0 {
		var err error
		payload, err = json.Marshal(online)
		if err != nil {
			return nil, err
		}
	}
	if ttl <= 0 {
		ttl = time.Minute
	}
	keys := []string{
		onlineKey(protocol, vin),
		onlineStateKey(protocol, vin),
		"vehicle:last_seen",
		realtimeIndexKey(protocol),
	}
	args := guardedOnlineStatusArgs(online, protocol, state, payload, ttl)
	return evaler.Eval(ctx, guardedOnlineStatusScript, keys, args...), nil
}

func guardedOnlineStatusArgs(online OnlineStatus, protocol envelope.Protocol, state map[string]any, payload []byte, ttl time.Duration) []any {
	args := []any{
		strconv.FormatInt(online.LastSeenMS, 10),
		strconv.FormatInt(ttl.Milliseconds(), 10),
		string(payload),
		onlineMember(protocol, online.VIN),
		strings.TrimSpace(online.VIN),
	}
	return appendMapPairs(args, state)
}

func guardedRealtimeKVArgs(eventTimeMS int64, values map[string]any, types map[string]any, meta map[string]any) []any {
	args := []any{strconv.FormatInt(eventTimeMS, 10)}
	args = appendMapPairs(args, values)
	args = appendMapPairs(args, types)
	args = appendMapPairs(args, meta)
	return args
}

func appendMapPairs(args []any, values map[string]any) []any {
	keys := make([]string, 0, len(values))
	for key := range values {
		key = strings.TrimSpace(key)
		if key == "" {
			continue
		}
		keys = append(keys, key)
	}
	sort.Strings(keys)
	args = append(args, strconv.Itoa(len(keys)))
	for _, key := range keys {
		args = append(args, key, fmt.Sprint(values[key]))
	}
	return args
}

func realtimeKVMapsForEnvelope(env envelope.FrameEnvelope) (map[string]any, map[string]any) {
	if fields, fieldTypes, ok := ParsedFieldsForEnvelope(env); ok {
		values := make(map[string]any, len(fields))
		types := make(map[string]any, len(fields))
		for field, value := range fields {
			if strings.TrimSpace(field) == "" {
				continue
			}
			if isRealtimeTotalMileageField(field) && !positiveNumber(value) {
				continue
			}
			stringValue, valueType, ok := stringifyKVValue(value)
			if !ok {
				continue
			}
			values[field] = stringValue
			if typed := strings.TrimSpace(fieldTypes[field]); typed != "" {
				types[field] = typed
			} else {
				types[field] = valueType
			}
		}
		return values, types
	}
	return nil, nil
}

func realtimeKVMaps(rows []RealtimeKVField) (map[string]any, map[string]any) {
	values := make(map[string]any, len(rows))
	types := make(map[string]any, len(rows))
	for _, row := range rows {
		field := realtimeKVFieldPath(row.Domain, row.Field)
		if field == "" {
			continue
		}
		values[field] = row.Value
		types[field] = row.ValueType
	}
	return values, types
}

func (r *Repository) setOnline(ctx context.Context, vehicleKey string, vin string, protocol envelope.Protocol, lastSeenMS int64, sourceEndpoint string) error {
	if protocol == "" {
		return nil
	}
	online := OnlineStatus{
		VehicleKey:     vehicleKey,
		VIN:            vin,
		Protocol:       protocol,
		Online:         true,
		LastSeenMS:     lastSeenMS,
		OfflineAfterMS: lastSeenMS + r.cfg.ttl().Milliseconds(),
		SourceEndpoint: sourceEndpoint,
		Protocols:      []envelope.Protocol{protocol},
		TTLSeconds:     int64(r.cfg.ttl().Seconds()),
	}
	return r.setOnlineStatus(ctx, online)
}

func (r *Repository) setOnlineStatus(ctx context.Context, online OnlineStatus) error {
	protocol := online.Protocol
	if protocol == "" && len(online.Protocols) > 0 {
		protocol = online.Protocols[0]
	}
	if protocol == "" {
		return nil
	}
	if online.OfflineAfterMS <= 0 {
		online.OfflineAfterMS = online.LastSeenMS + r.cfg.ttl().Milliseconds()
	}
	state := map[string]any{
		"vehicle_key":      online.VehicleKey,
		"vin":              online.VIN,
		"protocol":         string(protocol),
		"online":           strconv.FormatBool(true),
		"last_seen_ms":     strconv.FormatInt(online.LastSeenMS, 10),
		"offline_after_ms": strconv.FormatInt(online.OfflineAfterMS, 10),
		"ttl_seconds":      strconv.FormatInt(int64(r.cfg.ttl().Seconds()), 10),
		"source_endpoint":  online.SourceEndpoint,
	}
	pipe := r.client.Pipeline()
	payload, err := json.Marshal(online)
	if err != nil {
		return err
	}
	if _, err := evalGuardedOnlineStatus(ctx, pipe, online, state, payload, r.cfg.ttl()); err != nil {
		return err
	}
	_, err = pipe.Exec(ctx)
	return err
}

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 realtimeKVValuesKey(protocol envelope.Protocol, vin string) string {
	return "vehicle:rt-kv:" + string(protocol) + ":" + strings.TrimSpace(vin) + ":values"
}

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

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

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

func realtimeKVFieldPath(domain string, field string) string {
	domain = strings.TrimSpace(domain)
	field = strings.TrimSpace(field)
	switch {
	case domain == "":
		return field
	case field == "":
		return domain
	default:
		return domain + "." + field
	}
}

func eventTimeOrReceivedMS(env envelope.FrameEnvelope) int64 {
	eventMS, _ := envelope.NormalizedEventTimeMS(env)
	return eventMS
}

func onlineKey(protocol envelope.Protocol, vin string) string {
	return "vehicle:online:" + string(protocol) + ":" + strings.TrimSpace(vin)
}

func onlineStateKey(protocol envelope.Protocol, vin string) string {
	return "vehicle:online-state:" + string(protocol) + ":" + strings.TrimSpace(vin)
}

func onlineMember(protocol envelope.Protocol, vin string) string {
	return string(protocol) + ":" + strings.TrimSpace(vin)
}

func realtimeIndexKey(protocol envelope.Protocol) string {
	return "vehicle:rt-index:" + string(protocol)
}

func knownRealtimeProtocols() []envelope.Protocol {
	return []envelope.Protocol{envelope.ProtocolGB32960, envelope.ProtocolJT808, envelope.ProtocolYutongMQTT}
}

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

func normalizeOnlineListQuery(query OnlineListQuery) OnlineListQuery {
	query.Protocol = envelope.Protocol(strings.ToUpper(strings.TrimSpace(string(query.Protocol))))
	if query.Limit <= 0 {
		query.Limit = 100
	}
	if query.Limit > 1000 {
		query.Limit = 1000
	}
	if query.Offset < 0 {
		query.Offset = 0
	}
	return query
}

func splitOnlineMember(member string) (envelope.Protocol, string, bool) {
	protocolText, vin, ok := strings.Cut(strings.TrimSpace(member), ":")
	if !ok || strings.TrimSpace(protocolText) == "" || strings.TrimSpace(vin) == "" {
		return "", "", false
	}
	return envelope.Protocol(protocolText), strings.TrimSpace(vin), true
}

func parseInt64(value string) int64 {
	parsed, _ := strconv.ParseInt(strings.TrimSpace(value), 10, 64)
	return parsed
}

func firstPositiveInt64(values ...int64) int64 {
	for _, value := range values {
		if value > 0 {
			return value
		}
	}
	return 0
}

func firstNonEmptyString(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return strings.TrimSpace(value)
		}
	}
	return ""
}