feat: ingest OSM ways as places alongside nodes

cmd/ingestion/README.md

@@ -8,7 +8,7 @@ All configuration comes from environment variables.
 
 | Variable | Default | Notes |
 |---|---|---|
-| `OSM_PBF_PATH` | — | Required. Path to the `.osm.pbf` file. |
+| `OSM_PBF_PATH` | — | Required. Path to the `.osm.pbf` file. Must be preprocessed with `osmium add-locations-to-ways` so way geometry can be resolved — see `internal/sources/osm/README.md`. |
 | `DB_HOST` | `localhost` | |
 | `DB_PORT` | `5432` | |
 | `DB_USER` | `postgres` | |

cmd/ingestion/main_integration_test.go

@@ -54,10 +54,10 @@ func truncate(t *testing.T) {
 
 const fixturePBFPath = "../../testdata/andorra-sample.osm.pbf"
 
-// expectedPOICount is the number of nodes in the Andorra fixture that pass
+// expectedPOICount is the number of nodes and ways in the Andorra fixture that pass
 // osm.Evaluate. Locked in by inspecting the fixture; if the filter rules
 // change or the fixture is replaced, this number must be updated.
-const expectedPOICount = 976
+const expectedPOICount = 1370
 
 // pinned is a known POI from the Andorra fixture used to verify that the
 // transform -> upsert pipeline produces the expected place row.
@@ -188,6 +188,28 @@ func TestFullImport_AndorraFixture(t *testing.T) {
 		})
 	}
 
+	t.Run("way-derived place upserts with osm_type=way and no parent", func(t *testing.T) {
+		const wayID = 54858980
+		var got models.Place
+		err := db.Where("osm_id = ? AND osm_type = ?", wayID, models.OSMWay).First(&got).Error
+		if err != nil {
+			t.Fatalf("lookup osm_id=%d osm_type=way: %v", wayID, err)
+		}
+		if got.Name != "Olympia Centre" {
+			t.Errorf("name = %q, want %q", got.Name, "Olympia Centre")
+		}
+		if got.Category != models.CategoryShop {
+			t.Errorf("category = %q, want %q", got.Category, models.CategoryShop)
+		}
+		if got.ParentID != nil {
+			t.Errorf("parent_id = %v, want nil", *got.ParentID)
+		}
+		wantExternalID := fmt.Sprintf("way/%d", wayID)
+		if got.ExternalIDs["osm"].ID != wantExternalID {
+			t.Errorf("external_ids[osm].id = %q, want %q", got.ExternalIDs["osm"].ID, wantExternalID)
+		}
+	})
+
 	t.Run("re-import is idempotent on row count", func(t *testing.T) {
 		var before int64
 		if err := db.Model(&models.Place{}).Count(&before).Error; err != nil {

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/jackc/pgx/v5 v5.10.0
 	github.com/oapi-codegen/nethttp-middleware v1.1.2
 	github.com/oapi-codegen/runtime v1.4.1
+	github.com/paulmach/orb v0.12.0
 	github.com/paulmach/osm v0.9.0
 	github.com/testcontainers/testcontainers-go v0.42.0
 	github.com/testcontainers/testcontainers-go/modules/postgres v0.42.0
@@ -66,7 +67,6 @@ require (
 	github.com/oasdiff/yaml3 v0.0.14 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.1.1 // indirect
-	github.com/paulmach/orb v0.12.0 // indirect
 	github.com/paulmach/protoscan v0.2.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect

go.sum

@@ -48,8 +48,6 @@ github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/
 github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
-github.com/getkin/kin-openapi v0.140.0 h1:JFn675aXRFjyiZKa/BFWploGldQlI0gobp4J5k0EZ2g=
-github.com/getkin/kin-openapi v0.140.0/go.mod h1:lISrB64F0CPcuDJ3LdtPTMJBY8VENjR9wJBdrcT6J3g=
 github.com/getkin/kin-openapi v0.144.0 h1:hIRcTH+KjLfkLpYU6bSSfdFpi0fZi1fp+hSPi4aQu9Y=
 github.com/getkin/kin-openapi v0.144.0/go.mod h1:3BH9M9XDe/y9M5DSvEocVYAYq1w0qrhJHjC/vZi0AaY=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
@@ -140,12 +138,8 @@ github.com/oapi-codegen/nullable v1.1.0 h1:eAh8JVc5430VtYVnq00Hrbpag9PFRGWLjxR1/
 github.com/oapi-codegen/nullable v1.1.0/go.mod h1:KUZ3vUzkmEKY90ksAmit2+5juDIhIZhfDl+0PwOQlFY=
 github.com/oapi-codegen/runtime v1.4.1 h1:9nwLoI+KrWxzbBcp0jO/R8uXqbik/HUyCvPeU68Y/qo=
 github.com/oapi-codegen/runtime v1.4.1/go.mod h1:GwV7hC2hviaMzj+ITfHVRESK5J2W/GefVwIND/bMGvU=
-github.com/oasdiff/yaml v0.1.0 h1:0bqZjfKc/8S9urj4JuwepX41WX9EoA6ifhU3SV06cXg=
-github.com/oasdiff/yaml v0.1.0/go.mod h1:kOlRmMdL2X3vucLCEQO5u61SU22RysnfXvcttrZA1O0=
 github.com/oasdiff/yaml v0.1.1 h1:6nHx+pn9gBRM6YpBlFZFQGCCd1nuvqOBtTD3KKTgGxY=
 github.com/oasdiff/yaml v0.1.1/go.mod h1:EYJNoyktvWMJ0Hmhx+6qTaqMOsalUaRGT8Sj1hNcegU=
-github.com/oasdiff/yaml3 v0.0.13 h1:06svmvOHOVBqF81+sY2EUScvUI/iS/vl2VIeUUxZQwg=
-github.com/oasdiff/yaml3 v0.0.13/go.mod h1:y5+oSEHCPT/DGrS++Wc/479ERge0zTFxaF8PbGKcg2o=
 github.com/oasdiff/yaml3 v0.0.14 h1:aLJee3hxBK2H5wdXd9iPcIXb93Nty1Ge0pT171eHtkw=
 github.com/oasdiff/yaml3 v0.0.14/go.mod h1:csto2xfDjYccdUn/yw/bPjj/cYTdp6HtFA0J4TWG+gg=
 github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
@@ -233,8 +227,6 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
-golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
 golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -255,8 +247,6 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
-golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
 golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus=
 golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM=
 golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=

internal/sources/osm/README.md

@@ -1,13 +1,27 @@
 # internal/sources/osm
 
-Canonical source for the place registry. Reads an OpenStreetMap `.osm.pbf` file, filters nodes whose tags qualify them as POIs, transforms each into a `models.Place`, and streams them to the batcher in `cmd/ingestion`. Declares `SourceKindCanonical` so the dispatcher routes it through the canonical pipeline.
+Canonical source for the place registry. Reads an OpenStreetMap `.osm.pbf` file, filters nodes and ways whose tags qualify them as POIs, transforms each into a `models.Place`, and streams them to the batcher in `cmd/ingestion`. Declares `SourceKindCanonical` so the dispatcher routes it through the canonical pipeline.
+
+## Required preprocessing: annotated ways
+
+Raw PBF ways only store node ID references, not coordinates. Computing a way's shape requires resolving those references against actual node locations, which this package does not do itself. The input `.osm.pbf` **must** be preprocessed with [`osmium-tool`](https://osmcode.org/osmium-tool/) before ingestion:
+
+```
+osmium add-locations-to-ways <extract>.pbf -o <extract>-annotated.pbf
+```
+
+`OSM_PBF_PATH` (see `cmd/ingestion/README.md`) must point at the annotated output. Ways whose nodes aren't annotated produce an empty resolved geometry and are skipped with a warning (see Error handling) — ingestion does not detect or work around a missing preprocessing step beyond that.
+
+This is an additional flag on `osmium extract`, the step already required to scope a country/planet-wide extract down to a city or canton — not a new dependency.
 
 ## Pipeline
 
 ```mermaid
 flowchart LR
-  A[.osm.pbf file] --> B[StreamNodes<br/>paulmach/osm scanner]
-  B --> C[Evaluate tags]
+  A[.osm.pbf file] --> B[StreamElements<br/>paulmach/osm scanner]
+  B -- node --> C[Evaluate tags]
+  B -- way --> W[LineString + centroid]
+  W --> C
   C -- excluded --> D[Skip]
   C -- category, true --> E[TransformNode]
   E --> F[DeriveRank]
@@ -16,7 +30,9 @@ flowchart LR
   G --> H
 ```
 
-`StreamNodes` decodes the PBF and emits one OSM node at a time. Only matched POIs reach the sink.
+`StreamElements` decodes the PBF and emits one OSM node or way at a time; relations are skipped. Only matched POIs reach the sink.
+
+A way's representative `Lat`/`Lng` is the centroid (mean of coordinates) of its resolved `orb.LineString` — no polygon is persisted. Way-derived places are flat and independent, exactly like node-derived places: this package does not link a way (e.g. a mall) to the nodes inside it (e.g. its shops).
 
 ## Tag filtering: allowlist by design
 
@@ -35,7 +51,7 @@ Anything else is dropped.
 
 `TransformNode` builds a `models.Place` from an OSM node and a matched category. Coordinates come from the node; `Tags` is the full OSM tag map preserved as JSONB so `addr:street` and `addr:housenumber` (and anything else) remain available later — the identity matcher reads these tags directly when scoring address overlap.
 
-The natural key for upserts is `(osm_id, osm_type)`, where `osm_type` is `node`, `way`, or `relation`. Today only nodes are streamed.
+The natural key for upserts is `(osm_id, osm_type)`, where `osm_type` is `node`, `way`, or `relation`. Nodes and ways are streamed; relations (multipolygon buildings) are not handled.
 
 ## Accessibility tag mapping (v1)
 
@@ -66,6 +82,10 @@ There is no conflict detection anywhere in this mapping or downstream. The `whee
 
 Clients use the rank to prioritise results at low zoom levels — only landmarks at world view, establishments as you zoom in.
 
+## Error handling
+
+A way whose nodes aren't annotated (preprocessing step skipped, or the extract doesn't include all referenced nodes) produces an empty resolved `LineString`. `StreamElements` skips the way, logs a warning via `slog.Warn`, and does not fail the run.
+
 ## Dependency
 
-PBF decoding uses [`paulmach/osm`](https://github.com/paulmach/osm). The package wraps it just enough to stream nodes through `Sink` with a context for cancellation.
+PBF decoding uses [`paulmach/osm`](https://github.com/paulmach/osm); way geometry resolution uses its `paulmach/orb` dependency. The package wraps these just enough to stream nodes and ways through `Sink` with a context for cancellation.

internal/sources/osm/pbf.go

@@ -9,7 +9,9 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"log/slog"
 
+	"github.com/paulmach/orb"
 	pmosm "github.com/paulmach/osm"
 	"github.com/paulmach/osm/osmpbf"
 )
@@ -23,31 +25,61 @@ type Node struct {
 }
 
 // NodeSink is the callback invoked for each node read from a PBF stream.
-// Returning a non-nil error stops the stream and propagates the error to StreamNodes' caller.
+// Returning a non-nil error stops the stream and propagates the error to the caller.
 type NodeSink func(Node) error
 
-// StreamNodes reads a .osm.pbf stream and invokes sink for each node element.
-// Ways and relations are skipped silently.
-func StreamNodes(ctx context.Context, r io.Reader, sink NodeSink) error {
+// Way is a domain representation of an OSM way. Lat/Lng is the centroid of its resolved
+// nodes, not a stored polygon.
+type Way struct {
+	ID   int64
+	Lat  float64
+	Lng  float64
+	Tags map[string]string
+}
+
+// WaySink is the callback invoked for each way read from a PBF stream, sibling to NodeSink.
+type WaySink func(Way) error
+
+// StreamElements reads a .osm.pbf stream and invokes nodeSink per node and waySink per way
+// with resolved node locations. A nil waySink skips ways silently. Relations are always skipped.
+func StreamElements(ctx context.Context, r io.Reader, nodeSink NodeSink, waySink WaySink) error {
 	scanner := osmpbf.New(ctx, r, 1)
 	defer scanner.Close() //nolint:errcheck
 
 	for scanner.Scan() {
 		obj := scanner.Object()
-		node, ok := obj.(*pmosm.Node)
-		if !ok {
-			continue // skip ways, relations
-		}
-
-		domainNode := Node{
-			ID:   int64(node.ID),
-			Lat:  node.Lat,
-			Lng:  node.Lon,
-			Tags: node.Tags.Map(),
-		}
-
-		if err := sink(domainNode); err != nil {
-			return fmt.Errorf("node %d: %w", domainNode.ID, err)
+		switch o := obj.(type) {
+		case *pmosm.Node:
+			domainNode := Node{
+				ID:   int64(o.ID),
+				Lat:  o.Lat,
+				Lng:  o.Lon,
+				Tags: o.Tags.Map(),
+			}
+			if err := nodeSink(domainNode); err != nil {
+				return fmt.Errorf("node %d: %w", domainNode.ID, err)
+			}
+		case *pmosm.Way:
+			if waySink == nil {
+				continue
+			}
+			ls := o.LineString()
+			if len(ls) == 0 { // way nodes weren't annotated with locations in the PBF
+				slog.Warn("skipping way: no resolved node locations", "source", "osm", "way_id", o.ID)
+				continue
+			}
+			lat, lng := centroid(ls)
+			domainWay := Way{
+				ID:   int64(o.ID),
+				Lat:  lat,
+				Lng:  lng,
+				Tags: o.Tags.Map(),
+			}
+			if err := waySink(domainWay); err != nil {
+				return fmt.Errorf("way %d: %w", domainWay.ID, err)
+			}
+		default:
+			continue // skip relations
 		}
 	}
 
@@ -56,3 +88,13 @@ func StreamNodes(ctx context.Context, r io.Reader, sink NodeSink) error {
 	}
 	return nil
 }
+
+func centroid(ls orb.LineString) (lat, lng float64) {
+	var sumLat, sumLng float64
+	for _, p := range ls {
+		sumLng += p.X()
+		sumLat += p.Y()
+	}
+	n := float64(len(ls))
+	return sumLat / n, sumLng / n
+}

internal/sources/osm/pbf_test.go

@@ -10,9 +10,11 @@ import (
 	"errors"
 	"os"
 	"testing"
+
+	"github.com/paulmach/orb"
 )
 
-func TestStreamNodes_FixturePBF(t *testing.T) {
+func TestStreamElements_FixturePBF(t *testing.T) {
 	t.Parallel()
 	f, err := os.Open("../../../testdata/andorra-sample.osm.pbf")
 	if err != nil {
@@ -22,15 +24,15 @@ func TestStreamNodes_FixturePBF(t *testing.T) {
 
 	ctx := context.Background()
 	var totalNodes, includedPois int
-	err = StreamNodes(ctx, f, func(node Node) error {
+	err = StreamElements(ctx, f, func(node Node) error {
 		totalNodes++
 		if _, ok := Evaluate(node.Tags); ok {
 			includedPois++
 		}
 		return nil
-	})
+	}, nil)
 	if err != nil {
-		t.Fatalf("StreamNodes returned an error: %v", err)
+		t.Fatalf("StreamElements returned an error: %v", err)
 	}
 
 	if totalNodes == 0 {
@@ -39,7 +41,50 @@ func TestStreamNodes_FixturePBF(t *testing.T) {
 	t.Logf("streamed %d total nodes, %d qualifying POIs", totalNodes, includedPois)
 }
 
-func TestStreamNodes_StopsOnSinkError(t *testing.T) {
+func TestStreamElements_VisitsWays(t *testing.T) {
+	t.Parallel()
+	f, err := os.Open("../../../testdata/andorra-sample.osm.pbf")
+	if err != nil {
+		t.Skipf("fixture PBF not available: %v", err)
+	}
+	defer f.Close() //nolint:errcheck
+
+	ctx := context.Background()
+	var totalWays, resolvedWays int
+	err = StreamElements(ctx, f, func(Node) error { return nil }, func(way Way) error {
+		totalWays++
+		if way.Lat != 0 || way.Lng != 0 {
+			resolvedWays++
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("StreamElements returned an error: %v", err)
+	}
+	if totalWays == 0 {
+		t.Fatal("expected at least one way from the fixture PBF, got zero")
+	}
+	if resolvedWays == 0 {
+		t.Fatal("expected at least one way with a resolved centroid (annotated node locations), got zero")
+	}
+}
+
+func TestCentroid_MeanOfPoints(t *testing.T) {
+	t.Parallel()
+	ls := orb.LineString{
+		{6.0, 46.0},
+		{8.0, 48.0},
+	}
+	lat, lng := centroid(ls)
+	if lat != 47.0 {
+		t.Errorf("lat: got %v want 47.0", lat)
+	}
+	if lng != 7.0 {
+		t.Errorf("lng: got %v want 7.0", lng)
+	}
+}
+
+func TestStreamElements_StopsOnSinkError(t *testing.T) {
 	t.Parallel()
 	f, err := os.Open("../../../testdata/andorra-sample.osm.pbf")
 	if err != nil {
@@ -50,10 +95,10 @@ func TestStreamNodes_StopsOnSinkError(t *testing.T) {
 	ctx := context.Background()
 	sentinel := errors.New("stop")
 	count := 0
-	err = StreamNodes(ctx, f, func(node Node) error {
+	err = StreamElements(ctx, f, func(node Node) error {
 		count++
 		return sentinel
-	})
+	}, nil)
 	if err == nil {
 		t.Fatal("expected an error, got nil")
 	}