diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index 7afc75d6..435f591a 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -58,7 +58,7 @@ func main() { mux.Handle("/swagger-ui/", realtime.NewSwaggerUIHandler()) mux.Handle("/api/realtime/vehicles/", realtime.NewHandler(repository)) closeStats := func() {} - var updater realtimeUpdater = repository + var updater realtimeUpdater = fastRealtimeUpdater{repository: repository} if dsn := strings.TrimSpace(os.Getenv("MYSQL_DSN")); dsn != "" { db, err := sql.Open("mysql", dsn) if err != nil { @@ -93,7 +93,7 @@ func main() { envInt("MYSQL_REALTIME_ASYNC_WORKERS", 4), ) } - updater = compositeRealtimeUpdater{primary: repository, secondary: mysqlUpdater} + updater = compositeRealtimeUpdater{primary: fastRealtimeUpdater{repository: repository}, secondary: mysqlUpdater} logger.Info("realtime mysql snapshot enabled", "table", "vehicle_realtime_snapshot", "plate_binding_table", bindingTable, "plate_cache_ttl_seconds", envInt("PLATE_CACHE_TTL_SECONDS", 600)) } mux.Handle("/api/stats/daily-metrics", stats.NewMetricHandler(stats.NewMetricRepository(db))) @@ -216,6 +216,14 @@ type realtimeUpdater interface { Update(context.Context, envelope.FrameEnvelope) error } +type fastRealtimeUpdater struct { + repository *realtime.Repository +} + +func (u fastRealtimeUpdater) Update(ctx context.Context, env envelope.FrameEnvelope) error { + return u.repository.FastUpdate(ctx, env) +} + type compositeRealtimeUpdater struct { primary realtimeUpdater secondary realtimeUpdater diff --git a/go/vehicle-gateway/internal/realtime/kv.go b/go/vehicle-gateway/internal/realtime/kv.go index ddc6c1cd..5bc12d72 100644 --- a/go/vehicle-gateway/internal/realtime/kv.go +++ b/go/vehicle-gateway/internal/realtime/kv.go @@ -109,9 +109,6 @@ func flattenKV(prefix string, value any, out map[string]any) { return } itemKey := strconv.Itoa(index) - if serial := strings.TrimSpace(strconvAny(itemMap["serial_no"])); serial != "" && serial != "" { - itemKey = serial - } next := itemKey if prefix != "" { next = prefix + "." + itemKey diff --git a/go/vehicle-gateway/internal/realtime/model.go b/go/vehicle-gateway/internal/realtime/model.go index da2a9488..57268766 100644 --- a/go/vehicle-gateway/internal/realtime/model.go +++ b/go/vehicle-gateway/internal/realtime/model.go @@ -26,12 +26,15 @@ func (s Snapshot) Lightweight() Snapshot { } type OnlineStatus struct { - VehicleKey string `json:"vehicle_key"` - VIN string `json:"vin"` - Online bool `json:"online"` - LastSeenMS int64 `json:"last_seen_ms"` - Protocols []envelope.Protocol `json:"protocols,omitempty"` - TTLSeconds int64 `json:"ttl_seconds"` + VehicleKey string `json:"vehicle_key"` + VIN string `json:"vin"` + Protocol envelope.Protocol `json:"protocol,omitempty"` + Online bool `json:"online"` + LastSeenMS int64 `json:"last_seen_ms"` + OfflineAfterMS int64 `json:"offline_after_ms"` + SourceEndpoint string `json:"source_endpoint,omitempty"` + Protocols []envelope.Protocol `json:"protocols,omitempty"` + TTLSeconds int64 `json:"ttl_seconds"` } type Config struct { diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index e8bfeab2..124d51ee 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -39,7 +39,7 @@ func (r *Repository) FastUpdate(ctx context.Context, env envelope.FrameEnvelope) 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) + return r.setOnline(ctx, vehicleKey, vin, env.Protocol, env.ReceivedAtMS, env.SourceEndpoint) } func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) error { @@ -134,6 +134,7 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err online := OnlineStatus{ VehicleKey: vehicleKey, VIN: vin, + Protocol: env.Protocol, Online: true, LastSeenMS: env.ReceivedAtMS, Protocols: protocols, @@ -142,7 +143,7 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err 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() + return nil } func (r *Repository) GetMerged(ctx context.Context, vehicleKey string) (Snapshot, error) { @@ -164,21 +165,31 @@ func (r *Repository) GetRealtimeRaw(ctx context.Context, vehicleKey string, prot 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 + 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 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 + return best, nil } func (r *Repository) getSnapshot(ctx context.Context, key string) (Snapshot, error) { @@ -203,49 +214,86 @@ func (r *Repository) setKV(ctx context.Context, vin string, env envelope.FrameEn if len(rows) == 0 { return nil } - grouped := make(map[string]map[string]any) + values := make(map[string]any, len(rows)) + types := make(map[string]any, len(rows)) 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 + field := realtimeKVFieldPath(row.Domain, row.Field) + if field == "" { + continue } - values[row.Field] = row.Value - values["_value_type:"+row.Field] = row.ValueType + values[field] = row.Value + types[field] = row.ValueType + } + 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, } pipe := r.client.Pipeline() - for key, values := range grouped { - pipe.HSet(ctx, key, values) - } + pipe.HSet(ctx, realtimeKVValuesKey(env.Protocol, vin), values) + pipe.HSet(ctx, realtimeKVTypesKey(env.Protocol, vin), types) + pipe.HSet(ctx, realtimeKVMetaKey(env.Protocol, vin), meta) _, err := pipe.Exec(ctx) return err } -func (r *Repository) setOnline(ctx context.Context, vehicleKey string, vin string, protocols []envelope.Protocol, lastSeenMS int64) error { +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, - Online: true, - LastSeenMS: lastSeenMS, - Protocols: protocols, - TTLSeconds: int64(r.cfg.ttl().Seconds()), + 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()), } - 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() + return r.setOnlineStatus(ctx, online) } func (r *Repository) setOnlineStatus(ctx context.Context, online OnlineStatus) error { - return r.setJSON(ctx, onlineKey(online.VehicleKey), online, r.cfg.ttl()) + 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() + } + member := onlineMember(protocol, online.VIN) + 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 + } + pipe.Set(ctx, onlineKey(protocol, online.VIN), payload, r.cfg.ttl()) + pipe.HSet(ctx, onlineStateKey(protocol, online.VIN), state) + pipe.ZAdd(ctx, "vehicle:last_seen", redis.Z{Score: float64(online.LastSeenMS), Member: member}) + pipe.SAdd(ctx, realtimeIndexKey(protocol), online.VIN) + _, err = pipe.Exec(ctx) + return err } func (r *Repository) addProtocol(ctx context.Context, vehicleKey string, protocol envelope.Protocol) ([]envelope.Protocol, error) { @@ -567,12 +615,56 @@ 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 realtimeKVValuesKey(protocol envelope.Protocol, vin string) string { + return "vehicle:rt-kv:" + string(protocol) + ":" + strings.TrimSpace(vin) + ":values" } -func onlineKey(vehicleKey string) string { - return "vehicle:online:" + strings.TrimSpace(vehicleKey) +func realtimeKVTypesKey(protocol envelope.Protocol, vin string) string { + return "vehicle:rt-kv:" + string(protocol) + ":" + strings.TrimSpace(vin) + ":types" +} + +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 { + if env.EventTimeMS > 0 { + return env.EventTimeMS + } + return env.ReceivedAtMS +} + +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 { diff --git a/go/vehicle-gateway/internal/realtime/repository_test.go b/go/vehicle-gateway/internal/realtime/repository_test.go index ad840ec2..fdee435c 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -341,17 +341,34 @@ func TestRepositoryWritesRealtimeKVHashes(t *testing.T) { t.Fatalf("Update() error = %v", err) } - dataUnitsKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:data_units").Result() + dataUnitsKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:values").Result() if err != nil { - t.Fatalf("data_units kv HGetAll error = %v", err) + t.Fatalf("values kv HGetAll error = %v", err) } - if dataUnitsKV["0.value.soc_percent"] != "88" || dataUnitsKV["_event_id"] == "" || dataUnitsKV["_event_time_ms"] != "1000" { + if dataUnitsKV["data_units.0.value.soc_percent"] != "88" { t.Fatalf("data_units kv = %#v", dataUnitsKV) } - if dataUnitsKV["1.value.summaries.0.stack_water_outlet_temp_c"] != "63" { + if dataUnitsKV["data_units.1.value.summaries.0.stack_water_outlet_temp_c"] != "63" { t.Fatalf("data_units stack kv = %#v", dataUnitsKV) } - if ttl := repo.client.TTL(ctx, "vehicle:rt-kv:GB32960:VIN001:data_units").Val(); ttl != -1 { + typeKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:types").Result() + if err != nil { + t.Fatalf("types kv HGetAll error = %v", err) + } + if typeKV["data_units.0.value.soc_percent"] != "number" || typeKV["data_units.1.name"] != "string" { + t.Fatalf("types kv = %#v", typeKV) + } + metaKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:meta").Result() + if err != nil { + t.Fatalf("meta kv HGetAll error = %v", err) + } + if metaKV["event_id"] == "" || metaKV["event_time_ms"] != "1000" || metaKV["protocol"] != "GB32960" { + t.Fatalf("meta kv = %#v", metaKV) + } + if repo.client.Exists(ctx, "vehicle:rt-kv:GB32960:VIN001:data_units").Val() != 0 { + t.Fatal("old domain kv key should not be written") + } + if ttl := repo.client.TTL(ctx, "vehicle:rt-kv:GB32960:VIN001:values").Val(); ttl != -1 { t.Fatalf("kv ttl should not expire, got %v", ttl) } } @@ -375,19 +392,39 @@ func TestRepositoryFastUpdateOnlyWritesPermanentKVAndMinuteOnline(t *testing.T) t.Fatalf("FastUpdate() error = %v", err) } - dataUnitsKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:data_units").Result() + dataUnitsKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:values").Result() if err != nil { - t.Fatalf("data_units kv HGetAll error = %v", err) + t.Fatalf("values kv HGetAll error = %v", err) } - if dataUnitsKV["0.value.soc_percent"] != "88" || dataUnitsKV["_event_time_ms"] != "1000" { + if dataUnitsKV["data_units.0.value.soc_percent"] != "88" { t.Fatalf("data_units kv = %#v", dataUnitsKV) } - if ttl := repo.client.TTL(ctx, "vehicle:rt-kv:GB32960:VIN001:data_units").Val(); ttl != -1 { + typeKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:types").Result() + if err != nil { + t.Fatalf("types kv HGetAll error = %v", err) + } + if typeKV["data_units.0.value.soc_percent"] != "number" { + t.Fatalf("types kv = %#v", typeKV) + } + if ttl := repo.client.TTL(ctx, "vehicle:rt-kv:GB32960:VIN001:values").Val(); ttl != -1 { t.Fatalf("kv ttl should not expire, got %v", ttl) } - if ttl := repo.client.TTL(ctx, "vehicle:online:VIN001").Val(); ttl <= 0 || ttl > time.Minute { + if ttl := repo.client.TTL(ctx, "vehicle:online:GB32960:VIN001").Val(); ttl <= 0 || ttl > time.Minute { t.Fatalf("online ttl = %v, want within 1 minute", ttl) } + state, err := repo.client.HGetAll(ctx, "vehicle:online-state:GB32960:VIN001").Result() + if err != nil { + t.Fatalf("online state HGetAll error = %v", err) + } + if state["online"] != "true" || state["last_seen_ms"] != "1100" || state["offline_after_ms"] != "61100" { + t.Fatalf("online state = %#v", state) + } + if repo.client.ZScore(ctx, "vehicle:last_seen", "GB32960:VIN001").Err() != nil { + t.Fatal("last_seen should index protocol+vin") + } + if repo.client.SIsMember(ctx, "vehicle:rt-index:GB32960", "VIN001").Val() != true { + t.Fatal("rt-index should contain vin by protocol") + } if repo.client.Exists(ctx, "vehicle:latest:VIN001").Val() != 0 { t.Fatal("fast update should not write merged snapshot") }