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) Update(ctx context.Context, env envelope.FrameEnvelope) error { vin := strings.TrimSpace(env.VIN) 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, SourceEndpoint: env.SourceEndpoint, 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 } if existingProtocol.VehicleKey != "" { protocolSnapshot = existingProtocol if protocolSnapshot.VIN == "" && vin != "" { protocolSnapshot.VIN = vin } mergeFields(&protocolSnapshot, env.Fields, eventMS) protocolSnapshot.Parsed = mergeParsedForProtocol(env.Protocol, protocolSnapshot.Parsed, env.Parsed) if eventMS >= protocolSnapshot.EventTimeMS { protocolSnapshot.EventTimeMS = eventMS protocolSnapshot.EventID = env.StableEventID() protocolSnapshot.ReceivedAtMS = env.ReceivedAtMS protocolSnapshot.SourceEndpoint = env.SourceEndpoint } protocolSnapshot.UpdatedAtMS = nowMS } if err := r.setJSON(ctx, protocolKey(vehicleKey, env.Protocol), protocolSnapshot, 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 } 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{}, ProtocolData: map[envelope.Protocol]map[string]any{}, } } 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.SourceEndpoint = env.SourceEndpoint } merged.Protocols = protocols if merged.ProtocolData == nil { merged.ProtocolData = map[envelope.Protocol]map[string]any{} } merged.ProtocolData[env.Protocol] = cloneMap(protocolSnapshot.Parsed) 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.setJSON(ctx, onlineKey(vehicleKey), online, r.cfg.ttl()); 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) 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]) } 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 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 onlineKey(vehicleKey string) string { return "vehicle:online:" + strings.TrimSpace(vehicleKey) } func protocolsKey(vehicleKey string) string { return "vehicle:protocols:" + strings.TrimSpace(vehicleKey) }