diff --git a/go/vehicle-gateway/internal/realtime/model.go b/go/vehicle-gateway/internal/realtime/model.go index 6753015f..a60a4b9d 100644 --- a/go/vehicle-gateway/internal/realtime/model.go +++ b/go/vehicle-gateway/internal/realtime/model.go @@ -7,6 +7,7 @@ import ( ) type Snapshot struct { + VehicleKey string `json:"vehicle_key"` VIN string `json:"vin"` Protocol envelope.Protocol `json:"protocol,omitempty"` EventID string `json:"event_id,omitempty"` @@ -19,6 +20,7 @@ type Snapshot struct { } type OnlineStatus struct { + VehicleKey string `json:"vehicle_key"` VIN string `json:"vin"` Online bool `json:"online"` LastSeenMS int64 `json:"last_seen_ms"` diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index 1b1eaba5..1c2ea5d2 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -28,7 +28,8 @@ func NewRepository(client *redis.Client, cfg Config) *Repository { func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) error { vin := strings.TrimSpace(env.VIN) - if vin == "" { + vehicleKey := strings.TrimSpace(env.VehicleKey()) + if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") { return nil } nowMS := time.Now().UnixMilli() @@ -37,6 +38,7 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err eventMS = env.ReceivedAtMS } protocolSnapshot := Snapshot{ + VehicleKey: vehicleKey, VIN: vin, Protocol: env.Protocol, EventID: env.StableEventID(), @@ -47,16 +49,18 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err FieldTimesMS: fieldTimes(env.Fields, eventMS), UpdatedAtMS: nowMS, } - if err := r.setJSON(ctx, protocolKey(vin, env.Protocol), protocolSnapshot, r.cfg.ttl()); err != nil { + if err := r.setJSON(ctx, protocolKey(vehicleKey, env.Protocol), protocolSnapshot, r.cfg.ttl()); err != nil { return err } - merged, err := r.GetMerged(ctx, vin) + merged, err := r.GetMerged(ctx, vehicleKey) if err != nil && !errors.Is(err, redis.Nil) { return err } - if merged.VIN == "" { - merged = Snapshot{VIN: vin, Fields: map[string]any{}, FieldTimesMS: map[string]int64{}} + 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 { @@ -66,48 +70,50 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err merged.SourceEndpoint = env.SourceEndpoint } merged.UpdatedAtMS = nowMS - if err := r.setJSON(ctx, mergedKey(vin), merged, r.cfg.ttl()); err != nil { + if err := r.setJSON(ctx, mergedKey(vehicleKey), merged, r.cfg.ttl()); err != nil { return err } - protocols, err := r.addProtocol(ctx, vin, env.Protocol) + protocols, err := r.addProtocol(ctx, vehicleKey, env.Protocol) if 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(vin), online, r.cfg.ttl()); err != nil { + 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: vin}).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, vin string) (Snapshot, error) { - return r.getSnapshot(ctx, mergedKey(vin)) +func (r *Repository) GetMerged(ctx context.Context, vehicleKey string) (Snapshot, error) { + return r.getSnapshot(ctx, mergedKey(vehicleKey)) } -func (r *Repository) GetProtocol(ctx context.Context, vin string, protocol envelope.Protocol) (Snapshot, error) { - return r.getSnapshot(ctx, protocolKey(vin, protocol)) +func (r *Repository) GetProtocol(ctx context.Context, vehicleKey string, protocol envelope.Protocol) (Snapshot, error) { + return r.getSnapshot(ctx, protocolKey(vehicleKey, protocol)) } -func (r *Repository) IsOnline(ctx context.Context, vin string) (OnlineStatus, error) { +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(vin)).Bytes() + payload, err := r.client.Get(ctx, onlineKey(vehicleKey)).Bytes() if err != nil { if errors.Is(err, redis.Nil) { - return OnlineStatus{VIN: vin, Online: false}, 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(vin)).Val() + ttl := r.client.TTL(ctx, onlineKey(vehicleKey)).Val() status.Online = ttl > 0 status.TTLSeconds = int64(ttl.Seconds()) return status, nil @@ -130,8 +136,8 @@ func (r *Repository) setJSON(ctx context.Context, key string, value any, ttl tim return r.client.Set(ctx, key, payload, ttl).Err() } -func (r *Repository) addProtocol(ctx context.Context, vin string, protocol envelope.Protocol) ([]envelope.Protocol, error) { - key := protocolsKey(vin) +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 } @@ -207,18 +213,18 @@ func fieldTimes(fields map[string]any, eventMS int64) map[string]int64 { return out } -func mergedKey(vin string) string { - return "vehicle:latest:" + strings.TrimSpace(vin) +func mergedKey(vehicleKey string) string { + return "vehicle:latest:" + strings.TrimSpace(vehicleKey) } -func protocolKey(vin string, protocol envelope.Protocol) string { - return "vehicle:latest:" + strings.TrimSpace(vin) + ":" + string(protocol) +func protocolKey(vehicleKey string, protocol envelope.Protocol) string { + return "vehicle:latest:" + strings.TrimSpace(vehicleKey) + ":" + string(protocol) } -func onlineKey(vin string) string { - return "vehicle:online:" + strings.TrimSpace(vin) +func onlineKey(vehicleKey string) string { + return "vehicle:online:" + strings.TrimSpace(vehicleKey) } -func protocolsKey(vin string) string { - return "vehicle:protocols:" + strings.TrimSpace(vin) +func protocolsKey(vehicleKey string) string { + return "vehicle:protocols:" + strings.TrimSpace(vehicleKey) } diff --git a/go/vehicle-gateway/internal/realtime/repository_test.go b/go/vehicle-gateway/internal/realtime/repository_test.go index 880720ab..5497fc21 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -94,6 +94,44 @@ func TestRepositoryOnlineStatus(t *testing.T) { } } +func TestRepositoryUpdatesPhoneOnlyVehicleKey(t *testing.T) { + repo, closeFn := newTestRepository(t) + defer closeFn() + ctx := context.Background() + + if err := repo.Update(ctx, envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + Phone: "013307811170", + EventTimeMS: 1000, + ReceivedAtMS: 1100, + Fields: map[string]any{ + envelope.FieldSpeedKMH: 23.0, + envelope.FieldTotalMileageKM: 10003.7, + }, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + + vehicleKey := "JT808:013307811170" + merged, err := repo.GetMerged(ctx, vehicleKey) + if err != nil { + t.Fatalf("GetMerged() error = %v", err) + } + if merged.VehicleKey != vehicleKey || merged.VIN != "" { + t.Fatalf("unexpected identity: %#v", merged) + } + if merged.Fields[envelope.FieldTotalMileageKM] != 10003.7 { + t.Fatalf("merged fields = %#v", merged.Fields) + } + status, err := repo.IsOnline(ctx, vehicleKey) + if err != nil { + t.Fatalf("IsOnline() error = %v", err) + } + if !status.Online || status.VehicleKey != vehicleKey { + t.Fatalf("online status = %#v", status) + } +} + func TestRepositoryDoesNotOverwritePositiveMileageWithZero(t *testing.T) { repo, closeFn := newTestRepository(t) defer closeFn() @@ -159,6 +197,30 @@ func TestHandlerReturnsMergedSnapshot(t *testing.T) { } } +func TestHandlerReturnsPhoneOnlyVehicleKeySnapshot(t *testing.T) { + repo, closeFn := newTestRepository(t) + defer closeFn() + if err := repo.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + Phone: "013307811170", + EventTimeMS: 1000, + ReceivedAtMS: 1100, + Fields: map[string]any{envelope.FieldSpeedKMH: 12.3}, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + + req := httptest.NewRequest(http.MethodGet, "/api/realtime/vehicles/JT808:013307811170", nil) + rec := httptest.NewRecorder() + NewHandler(repo).ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + if !stringsContains(rec.Body.String(), `"vehicle_key":"JT808:013307811170"`) { + t.Fatalf("unexpected body: %s", rec.Body.String()) + } +} + func newTestRepository(t *testing.T) (*Repository, func()) { t.Helper() server, err := miniredis.Run()