refactor(go): simplify redis realtime projections

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

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 != "<nil>" {
-				itemKey = serial
-			}
 			next := itemKey
 			if prefix != "" {
 				next = prefix + "." + itemKey

go/vehicle-gateway/internal/realtime/model.go

@@ -26,12 +26,15 @@ func (s Snapshot) Lightweight() Snapshot {
 }
 
 type OnlineStatus struct {
-	VehicleKey string              `json:"vehicle_key"`
+	VehicleKey     string              `json:"vehicle_key"`
-	VIN        string              `json:"vin"`
+	VIN            string              `json:"vin"`
-	Online     bool                `json:"online"`
+	Protocol       envelope.Protocol   `json:"protocol,omitempty"`
-	LastSeenMS int64               `json:"last_seen_ms"`
+	Online         bool                `json:"online"`
-	Protocols  []envelope.Protocol `json:"protocols,omitempty"`
+	LastSeenMS     int64               `json:"last_seen_ms"`
-	TTLSeconds int64               `json:"ttl_seconds"`
+	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 {

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
+	best := OnlineStatus{VehicleKey: vehicleKey, VIN: vehicleKey, Online: false}
-	payload, err := r.client.Get(ctx, onlineKey(vehicleKey)).Bytes()
+	for _, protocol := range knownRealtimeProtocols() {
-	if err != nil {
+		key := onlineKey(protocol, vehicleKey)
-		if errors.Is(err, redis.Nil) {
+		payload, err := r.client.Get(ctx, key).Bytes()
-			return OnlineStatus{VehicleKey: vehicleKey, VIN: vehicleKey, Online: false}, nil
+		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 best, 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) {
@@ -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)
+		field := realtimeKVFieldPath(row.Domain, row.Field)
-		values := grouped[key]
+		if field == "" {
-		if values == nil {
+			continue
-			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[field] = row.Value
-		values["_value_type:"+row.Field] = row.ValueType
+		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, realtimeKVValuesKey(env.Protocol, vin), values)
-		pipe.HSet(ctx, key, 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,
+		VehicleKey:     vehicleKey,
-		VIN:        vin,
+		VIN:            vin,
-		Online:     true,
+		Protocol:       protocol,
-		LastSeenMS: lastSeenMS,
+		Online:         true,
-		Protocols:  protocols,
+		LastSeenMS:     lastSeenMS,
-		TTLSeconds: int64(r.cfg.ttl().Seconds()),
+		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 r.setOnlineStatus(ctx, online)
-		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())
+	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 {
+func realtimeKVValuesKey(protocol envelope.Protocol, vin string) string {
-	return "vehicle:rt-kv:" + string(protocol) + ":" + strings.TrimSpace(vin) + ":" + strings.TrimSpace(domain)
+	return "vehicle:rt-kv:" + string(protocol) + ":" + strings.TrimSpace(vin) + ":values"
 }
 
-func onlineKey(vehicleKey string) string {
+func realtimeKVTypesKey(protocol envelope.Protocol, vin string) string {
-	return "vehicle:online:" + strings.TrimSpace(vehicleKey)
+	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 {

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")
 	}