From a9974bf327bbbd6a17ed933f68f1648eb9953ce1 Mon Sep 17 00:00:00 2001 From: Jilles van Gurp Date: Tue, 9 Jun 2026 16:21:30 +0200 Subject: [PATCH] Add provider-based trackable auto-association --- docs/connectors-websocket.md | 9 +++ engineering/implementation-plan.md | 2 +- internal/hub/metadata_cache.go | 102 ++++++++++++++++++++-------- internal/hub/metadata_cache_test.go | 7 +- internal/hub/service.go | 63 +++++++++++++++-- internal/hub/service_test.go | 59 ++++++++++++++++ specifications/omlox/trackables.md | 6 ++ tests/integration/scenario_test.go | 61 +++++++++++++++++ 8 files changed, 273 insertions(+), 36 deletions(-) diff --git a/docs/connectors-websocket.md b/docs/connectors-websocket.md index 452c72b..baabe54 100644 --- a/docs/connectors-websocket.md +++ b/docs/connectors-websocket.md @@ -112,6 +112,15 @@ REST for idempotent metadata setup: - upsert `Trackable` resources when the upstream source has stable asset IDs - optionally create `Zone` and `Fence` resources before ingest starts +For trackable association, the current hub behavior supports two practical +patterns: + +- when the connector already knows the asset identity for an observation, send + `trackables` on each `location_updates` payload +- when one upstream provider stream is dedicated to one asset, store that + provider ID in `Trackable.location_providers`; the hub now auto-associates + incoming locations when exactly one trackable matches the `provider_id` + That split is usually the easiest connector design: - REST for durable metadata and bootstrap diff --git a/engineering/implementation-plan.md b/engineering/implementation-plan.md index d421092..a5a68a2 100644 --- a/engineering/implementation-plan.md +++ b/engineering/implementation-plan.md @@ -18,7 +18,7 @@ The repository documentation is now split by audience: software/runtime document - Core REST CRUD is implemented for zones, providers, trackables, and fences through a shared service layer backed by Postgres and `sqlc`. - Hub-generated UUIDs for REST-managed resources, derived fence and collision events, and JSON-RPC caller IDs now use UUIDv7 so newly issued identifiers are time-sortable. - UUID generation is intentionally centralized behind `internal/ids` so the repository can switch from `github.com/google/uuid` to a future Go standard-library UUIDv7 implementation with a narrowly scoped change once that support is available and stable. -- Provider ingestion endpoints are implemented for locations and proximities, with in-memory deduplication, latest-state tracking, proximity hysteresis, fence membership, and collision state. +- Provider ingestion endpoints are implemented for locations and proximities, with in-memory deduplication, latest-state tracking, unique-provider trackable auto-association from `Trackable.location_providers`, proximity hysteresis, fence membership, and collision state. - Ingest accepts omitted `crs`, `local`, and named EPSG codes, derives true local and WGS84 variants when transformation is possible, and suppresses only the unavailable output topic when it is not. - A shared internal event bus now fans normalized hub events out to MQTT and WebSocket consumers instead of keeping MQTT as the only outbound path. - MQTT is broker-backed and wired into startup, inbound ingest topics, and outbound location, fence-event, trackable-motion, and optional collision publication. diff --git a/internal/hub/metadata_cache.go b/internal/hub/metadata_cache.go index 90b8cfd..b95f70d 100644 --- a/internal/hub/metadata_cache.go +++ b/internal/hub/metadata_cache.go @@ -23,20 +23,21 @@ const ( ) type metadataSnapshot struct { - zones []gen.Zone - zonesByID map[string]gen.Zone - zonesByForeignID map[string]gen.Zone - zoneSignatures map[string]string - fences []gen.Fence - fencesByID map[string]gen.Fence - fenceSignatures map[string]string - fenceIndexes map[string]*fenceSpatialIndex - trackables []gen.Trackable - trackablesByID map[string]gen.Trackable - trackableSignatures map[string]string - providers []gen.LocationProvider - providersByID map[string]gen.LocationProvider - providerSignatures map[string]string + zones []gen.Zone + zonesByID map[string]gen.Zone + zonesByForeignID map[string]gen.Zone + zoneSignatures map[string]string + fences []gen.Fence + fencesByID map[string]gen.Fence + fenceSignatures map[string]string + fenceIndexes map[string]*fenceSpatialIndex + trackables []gen.Trackable + trackablesByID map[string]gen.Trackable + trackablesByProviderID map[string][]gen.Trackable + trackableSignatures map[string]string + providers []gen.LocationProvider + providersByID map[string]gen.LocationProvider + providerSignatures map[string]string } const ( @@ -184,20 +185,21 @@ type providerRecord struct { func newMetadataSnapshot(zones []zoneRecord, fences []fenceRecord, trackables []trackableRecord, providers []providerRecord) metadataSnapshot { snapshot := metadataSnapshot{ - zones: make([]gen.Zone, 0, len(zones)), - zonesByID: make(map[string]gen.Zone, len(zones)), - zonesByForeignID: make(map[string]gen.Zone, len(zones)), - zoneSignatures: make(map[string]string, len(zones)), - fences: make([]gen.Fence, 0, len(fences)), - fencesByID: make(map[string]gen.Fence, len(fences)), - fenceSignatures: make(map[string]string, len(fences)), - fenceIndexes: make(map[string]*fenceSpatialIndex), - trackables: make([]gen.Trackable, 0, len(trackables)), - trackablesByID: make(map[string]gen.Trackable, len(trackables)), - trackableSignatures: make(map[string]string, len(trackables)), - providers: make([]gen.LocationProvider, 0, len(providers)), - providersByID: make(map[string]gen.LocationProvider, len(providers)), - providerSignatures: make(map[string]string, len(providers)), + zones: make([]gen.Zone, 0, len(zones)), + zonesByID: make(map[string]gen.Zone, len(zones)), + zonesByForeignID: make(map[string]gen.Zone, len(zones)), + zoneSignatures: make(map[string]string, len(zones)), + fences: make([]gen.Fence, 0, len(fences)), + fencesByID: make(map[string]gen.Fence, len(fences)), + fenceSignatures: make(map[string]string, len(fences)), + fenceIndexes: make(map[string]*fenceSpatialIndex), + trackables: make([]gen.Trackable, 0, len(trackables)), + trackablesByID: make(map[string]gen.Trackable, len(trackables)), + trackablesByProviderID: make(map[string][]gen.Trackable), + trackableSignatures: make(map[string]string, len(trackables)), + providers: make([]gen.LocationProvider, 0, len(providers)), + providersByID: make(map[string]gen.LocationProvider, len(providers)), + providerSignatures: make(map[string]string, len(providers)), } for _, record := range zones { item := record.Zone @@ -218,6 +220,9 @@ func newMetadataSnapshot(zones []zoneRecord, fences []fenceRecord, trackables [] item := record.Trackable snapshot.trackables = append(snapshot.trackables, item) snapshot.trackablesByID[item.Id.String()] = item + for _, providerID := range stringSliceValue(item.LocationProviders) { + snapshot.trackablesByProviderID[providerID] = append(snapshot.trackablesByProviderID[providerID], item) + } snapshot.trackableSignatures[item.Id.String()] = record.Signature } for _, record := range providers { @@ -293,6 +298,12 @@ func (c *MetadataCache) TrackableByID(id string) (gen.Trackable, bool) { return item, ok } +func (c *MetadataCache) TrackablesByProviderID(id string) []gen.Trackable { + snapshot := c.current() + items := snapshot.trackablesByProviderID[id] + return append([]gen.Trackable(nil), items...) +} + func (c *MetadataCache) ProviderByID(id string) (gen.LocationProvider, bool) { snapshot := c.current() item, ok := snapshot.providersByID[id] @@ -372,8 +383,18 @@ func (c *MetadataCache) UpsertTrackable(item gen.Trackable, signature string) { next := c.snapshot next.trackables = upsertByTrackableID(next.trackables, item) next.trackablesByID = cloneTrackableMap(next.trackablesByID) + next.trackablesByProviderID = cloneTrackableSliceMap(next.trackablesByProviderID) next.trackableSignatures = cloneStringMap(next.trackableSignatures) next.trackablesByID[item.Id.String()] = item + for providerID, items := range next.trackablesByProviderID { + next.trackablesByProviderID[providerID] = removeTrackableFromSlice(items, item.Id) + if len(next.trackablesByProviderID[providerID]) == 0 { + delete(next.trackablesByProviderID, providerID) + } + } + for _, providerID := range stringSliceValue(item.LocationProviders) { + next.trackablesByProviderID[providerID] = append(next.trackablesByProviderID[providerID], item) + } next.trackableSignatures[item.Id.String()] = signature c.snapshot = next } @@ -384,8 +405,15 @@ func (c *MetadataCache) DeleteTrackable(id openapi_types.UUID) { next := c.snapshot next.trackables = removeTrackableByID(next.trackables, id) next.trackablesByID = cloneTrackableMap(next.trackablesByID) + next.trackablesByProviderID = cloneTrackableSliceMap(next.trackablesByProviderID) next.trackableSignatures = cloneStringMap(next.trackableSignatures) delete(next.trackablesByID, id.String()) + for providerID, items := range next.trackablesByProviderID { + next.trackablesByProviderID[providerID] = removeTrackableFromSlice(items, id) + if len(next.trackablesByProviderID[providerID]) == 0 { + delete(next.trackablesByProviderID, providerID) + } + } delete(next.trackableSignatures, id.String()) c.snapshot = next } @@ -741,6 +769,24 @@ func cloneTrackableMap(in map[string]gen.Trackable) map[string]gen.Trackable { return out } +func cloneTrackableSliceMap(in map[string][]gen.Trackable) map[string][]gen.Trackable { + out := make(map[string][]gen.Trackable, len(in)) + for key, value := range in { + out[key] = append([]gen.Trackable(nil), value...) + } + return out +} + +func removeTrackableFromSlice(items []gen.Trackable, id openapi_types.UUID) []gen.Trackable { + next := make([]gen.Trackable, 0, len(items)) + for _, item := range items { + if item.Id != id { + next = append(next, item) + } + } + return next +} + func cloneProviderMap(in map[string]gen.LocationProvider) map[string]gen.LocationProvider { out := make(map[string]gen.LocationProvider, len(in)) for key, value := range in { diff --git a/internal/hub/metadata_cache_test.go b/internal/hub/metadata_cache_test.go index 1f4ed5e..8f8280d 100644 --- a/internal/hub/metadata_cache_test.go +++ b/internal/hub/metadata_cache_test.go @@ -16,7 +16,8 @@ func TestNewMetadataCacheBuildsIndexes(t *testing.T) { t.Parallel() zone := testZoneWithForeignID(t, uuid.New(), "uwb", "foreign-zone", [2]float32{1, 2}, nil, nil) - trackable := gen.Trackable{Id: uuidAsOpenAPI(uuid.New()), Type: gen.TrackableTypeOmlox} + providerIDs := gen.StringIdList{"provider-a"} + trackable := gen.Trackable{Id: uuidAsOpenAPI(uuid.New()), Type: gen.TrackableTypeOmlox, LocationProviders: &providerIDs} provider := gen.LocationProvider{Id: "provider-a", Type: "uwb"} fence := testPointFence(t, uuid.New(), [2]float32{1, 2}, 5) @@ -39,6 +40,10 @@ func TestNewMetadataCacheBuildsIndexes(t *testing.T) { if _, ok := cache.TrackableByID(trackable.Id.String()); !ok { t.Fatal("expected trackable lookup by id") } + matches := cache.TrackablesByProviderID(provider.Id) + if len(matches) != 1 || matches[0].Id != trackable.Id { + t.Fatalf("expected trackable lookup by provider id, got %+v", matches) + } if _, ok := cache.ProviderByID(provider.Id); !ok { t.Fatal("expected provider lookup by id") } diff --git a/internal/hub/service.go b/internal/hub/service.go index a8348a6..737a25d 100644 --- a/internal/hub/service.go +++ b/internal/hub/service.go @@ -759,21 +759,25 @@ func (s *Service) DeleteFence(ctx context.Context, id openapi_types.UUID) error // updates. func (s *Service) ProcessLocations(ctx context.Context, locations []gen.Location) error { for _, location := range locations { + resolvedLocation, err := s.resolveLocationTrackables(ctx, location) + if err != nil { + return err + } itemCtx := observability.WithIngestTransport(ctx, observability.IngestTransportFromContext(ctx)) itemCtx, span := s.telemetry().StartSpan(itemCtx, "hub.ingest.location", - attribute.String("provider_id", location.ProviderId), - attribute.String("provider_type", location.ProviderType), - attribute.String("source", location.Source), - attribute.StringSlice("trackable_ids", stringSliceValue(location.Trackables)), + attribute.String("provider_id", resolvedLocation.ProviderId), + attribute.String("provider_type", resolvedLocation.ProviderType), + attribute.String("source", resolvedLocation.Source), + attribute.StringSlice("trackable_ids", stringSliceValue(resolvedLocation.Trackables)), ) start := time.Now() - if err := validateLocation(location); err != nil { + if err := validateLocation(resolvedLocation); err != nil { span.RecordError(err) s.telemetry().RecordIngestRecord(itemCtx, "location", "invalid") span.End() return err } - if err := s.recordLocation(itemCtx, location, s.cfg.LocationTTL); err != nil { + if err := s.recordLocation(itemCtx, resolvedLocation, s.cfg.LocationTTL); err != nil { span.RecordError(err) s.telemetry().RecordIngestRecord(itemCtx, "location", "failed") span.End() @@ -785,6 +789,53 @@ func (s *Service) ProcessLocations(ctx context.Context, locations []gen.Location return nil } +func (s *Service) resolveLocationTrackables(ctx context.Context, location gen.Location) (gen.Location, error) { + if location.Trackables != nil && len(*location.Trackables) > 0 { + return location, nil + } + providerID := strings.TrimSpace(location.ProviderId) + if providerID == "" { + return location, nil + } + + var matches []gen.Trackable + if cache := s.metadataCache(); cache != nil { + matches = cache.TrackablesByProviderID(providerID) + } else { + trackables, err := s.ListTrackables(ctx) + if err != nil { + return gen.Location{}, err + } + matches = trackablesForProvider(trackables, providerID) + } + if len(matches) != 1 { + return location, nil + } + + out, err := cloneLocation(location) + if err != nil { + return gen.Location{}, err + } + trackableIDs := gen.StringIdList{matches[0].Id.String()} + out.Trackables = &trackableIDs + associated := true + out.Associated = &associated + return out, nil +} + +func trackablesForProvider(trackables []gen.Trackable, providerID string) []gen.Trackable { + matches := make([]gen.Trackable, 0, 1) + for _, trackable := range trackables { + for _, candidate := range stringSliceValue(trackable.LocationProviders) { + if candidate == providerID { + matches = append(matches, trackable) + break + } + } + } + return matches +} + // ProcessProximities validates, resolves, and republishes provider proximity // updates. func (s *Service) ProcessProximities(ctx context.Context, proximities []gen.Proximity) error { diff --git a/internal/hub/service_test.go b/internal/hub/service_test.go index 8594812..3e5b5cc 100644 --- a/internal/hub/service_test.go +++ b/internal/hub/service_test.go @@ -151,6 +151,65 @@ func TestLocationPropertiesExcludeGeometryAndPreserveFields(t *testing.T) { } } +func TestResolveLocationTrackablesAssociatesUniqueProviderMatch(t *testing.T) { + t.Parallel() + + providerIDs := gen.StringIdList{"provider-a"} + trackable := gen.Trackable{ + Id: uuidAsOpenAPI(uuid.New()), + Type: gen.TrackableTypeOmlox, + LocationProviders: &providerIDs, + } + service := &Service{ + metadata: &MetadataCache{snapshot: newMetadataSnapshot(nil, nil, []trackableRecord{{Trackable: trackable}}, nil)}, + logger: zapTestLogger(t), + } + location := testLocation(t, nil) + + resolved, err := service.resolveLocationTrackables(context.Background(), location) + if err != nil { + t.Fatalf("resolveLocationTrackables failed: %v", err) + } + if resolved.Trackables == nil || len(*resolved.Trackables) != 1 || (*resolved.Trackables)[0] != trackable.Id.String() { + t.Fatalf("expected resolved trackable id %s, got %+v", trackable.Id.String(), resolved.Trackables) + } + if resolved.Associated == nil || !*resolved.Associated { + t.Fatalf("expected associated=true, got %+v", resolved.Associated) + } +} + +func TestResolveLocationTrackablesLeavesAmbiguousProviderUnassociated(t *testing.T) { + t.Parallel() + + providerIDs := gen.StringIdList{"provider-a"} + trackableA := gen.Trackable{ + Id: uuidAsOpenAPI(uuid.New()), + Type: gen.TrackableTypeOmlox, + LocationProviders: &providerIDs, + } + trackableB := gen.Trackable{ + Id: uuidAsOpenAPI(uuid.New()), + Type: gen.TrackableTypeOmlox, + LocationProviders: &providerIDs, + } + service := &Service{ + metadata: &MetadataCache{snapshot: newMetadataSnapshot(nil, nil, []trackableRecord{{Trackable: trackableA}, {Trackable: trackableB}}, nil)}, + logger: zapTestLogger(t), + } + location := testLocation(t, nil) + + resolved, err := service.resolveLocationTrackables(context.Background(), location) + if err != nil { + t.Fatalf("resolveLocationTrackables failed: %v", err) + } + if resolved.Trackables != nil { + t.Fatalf("expected ambiguous provider to remain unassociated, got %+v", resolved.Trackables) + } + if resolved.Associated != nil { + t.Fatalf("expected associated to remain unset, got %+v", resolved.Associated) + } +} + func TestFenceEventPropertiesExcludeGeometryAndPreserveFields(t *testing.T) { t.Parallel() diff --git a/specifications/omlox/trackables.md b/specifications/omlox/trackables.md index 3d6a66d..c4416e8 100644 --- a/specifications/omlox/trackables.md +++ b/specifications/omlox/trackables.md @@ -36,3 +36,9 @@ Trackable API is defined as a management API; companion OpenAPI is expected to d - A trackable location is based on updates from assigned location providers. - `omlox` type supports self-assignment style, `virtual` type supports API assignment. - Trackable/fence interaction and collision behavior is normative in chapters 9, 10, and 11. + +Current repository behavior: +- If an incoming `Location` already names `trackables`, the hub uses that explicit association. +- If `trackables` is omitted and exactly one stored `Trackable.location_providers` entry matches the incoming `provider_id`, the hub auto-associates the location to that trackable and marks it as associated. +- If no match or more than one match exists for the provider, the hub currently leaves the location unassociated rather than guessing. +- `locating_rules` are modeled in the contract but are not yet applied at runtime to resolve ambiguous multi-trackable provider matches. diff --git a/tests/integration/scenario_test.go b/tests/integration/scenario_test.go index 8f0c10b..2a6a401 100644 --- a/tests/integration/scenario_test.go +++ b/tests/integration/scenario_test.go @@ -197,6 +197,48 @@ func TestScenarioMixedRESTAndMQTTIngestShareOneHubSubscribers(t *testing.T) { waitForWebSocketProviders(t, wsConn, providers, 15*time.Second) } +func TestScenarioProviderAssignedTrackableAutoAssociatesLocation(t *testing.T) { + defer func() { + if r := recover(); r != nil { + t.Skipf("docker/testcontainers unavailable: %v", r) + } + }() + + _, appBaseURL, brokerURL := startHubNoAuth(t) + token := adminToken(t) + zoneID := createScenarioZone(t, appBaseURL, token) + providerID := scopedID(t, "auto-assoc-provider") + trackableID := createTrackableWithProviders(t, appBaseURL, token, scopedID(t, "auto-assoc-trackable"), []string{providerID}) + + subscriber, messages := mqttSubscriber(t, brokerURL, mqtt.TopicLocationLocal(providerID), mqtt.TopicTrackableMotionLocal(trackableID)) + defer subscriber.Disconnect(250) + + resp := requestJSON(t, http.MethodPost, appBaseURL+"/v2/providers/locations", token, []map[string]any{{ + "crs": "local", + "position": pointPayload(5, 7), + "provider_id": providerID, + "provider_type": "uwb", + "source": zoneID, + }}) + assertStatusAndClose(t, resp, http.StatusAccepted) + + location := waitForLocation(t, messages[mqtt.TopicLocationLocal(providerID)], 10*time.Second) + if location.Trackables == nil || len(*location.Trackables) != 1 || (*location.Trackables)[0] != trackableID { + t.Fatalf("expected auto-associated trackable %s, got %+v", trackableID, location.Trackables) + } + if location.Associated == nil || !*location.Associated { + t.Fatalf("expected associated=true, got %+v", location.Associated) + } + + motion := waitForTrackableMotionAtLocation(t, messages[mqtt.TopicTrackableMotionLocal(trackableID)], 5, 7, 10*time.Second) + if motion.Id != trackableID { + t.Fatalf("unexpected trackable motion id: got %s want %s", motion.Id, trackableID) + } + if motion.Location.Trackables == nil || len(*motion.Location.Trackables) != 1 || (*motion.Location.Trackables)[0] != trackableID { + t.Fatalf("expected motion location to carry auto-associated trackable %s, got %+v", trackableID, motion.Location.Trackables) + } +} + func TestScenarioTenObjectsTraverseFenceLayoutAndUpdateMotions(t *testing.T) { defer func() { if r := recover(); r != nil { @@ -298,6 +340,25 @@ func createTrackable(t *testing.T, appBaseURL, token, name string) string { return trackable.ID } +func createTrackableWithProviders(t *testing.T, appBaseURL, token, name string, providerIDs []string) string { + t.Helper() + + createResp := requestJSON(t, http.MethodPost, appBaseURL+"/v2/trackables", token, map[string]any{ + "type": "omlox", + "name": name, + "location_providers": providerIDs, + }) + assertStatus(t, createResp, http.StatusCreated) + var trackable struct { + ID string `json:"id"` + } + decodeResponse(t, createResp, &trackable) + if trackable.ID == "" { + t.Fatal("expected created trackable id") + } + return trackable.ID +} + func createLocalPointFence(t *testing.T, appBaseURL, token, zoneID, foreignID string, x, y, radius float64) scenarioFence { t.Helper() foreignID = scopedID(t, foreignID)