Add provider-based trackable auto-association

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

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.

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 {

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

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 {

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()
 

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.