Apply a stop consolidation file on static import

gtfsdb/helpers.go

@@ -34,6 +34,10 @@ type GtfsData struct {
 	Hash   string
 	Source string
 
+	// StopReplacements removes the stops a stop consolidation file absorbs,
+	// keyed by stop ID. It is nil when no file is configured.
+	StopReplacements StopReplacements
+
 	// stream reads the stop times and shape points, which Static lacks.
 	stream *gtfs.StaticStream
 }
@@ -372,12 +376,12 @@ func (c *Client) StoreGtfsData(ctx context.Context, data *GtfsData) (bool, error
 		//
 		// Future: If pathways or station accessibility features are needed, consider
 		// making lat/lon nullable in the schema and updating handlers accordingly.
-		if s.Latitude == nil || s.Longitude == nil {
+		if s.Latitude == nil || s.Longitude == nil || data.StopReplacements.isAbsorbed(s.Id) {
 			continue
 		}
 		parentStation := ""
 		if s.Parent != nil {
-			parentStation = s.Parent.Id
+			parentStation = data.StopReplacements.storedStopID(s.Parent.Id)
 		}
 		params := CreateStopParams{
 			ID: s.Id,
@@ -404,6 +408,9 @@ func (c *Client) StoreGtfsData(ctx context.Context, data *GtfsData) (bool, error
 	if err := c.bulkInsertStops(ctx, allStopParams, tx); err != nil {
 		return false, fmt.Errorf("unable to create stops: %w", err)
 	}
+	if err := storeStopConsolidations(ctx, qtx, data.Static.Stops, data.StopReplacements); err != nil {
+		return false, err
+	}
 
 	logging.LogOperation(logger, "agencies_and_routes_inserted",
 		slog.Int("agencies", len(data.Static.Agencies)),
@@ -736,6 +743,9 @@ func (c *Client) clearAllGTFSDataWithQueries(ctx context.Context, q *Queries) er
 	if err := q.ClearCalendar(ctx); err != nil {
 		return fmt.Errorf("error clearing calendar: %w", err)
 	}
+	if err := q.ClearStopConsolidations(ctx); err != nil {
+		return fmt.Errorf("error clearing stop_consolidations: %w", err)
+	}
 	if err := q.ClearStops(ctx); err != nil {
 		return fmt.Errorf("error clearing stops: %w", err)
 	}

gtfsdb/query.sql

@@ -1540,3 +1540,27 @@ FROM
     JOIN stops s ON s.id = st.stop_id
 GROUP BY
     r.agency_id;
+
+-- name: CreateStopConsolidation :exec
+INSERT INTO
+    stop_consolidations (
+        absorbed_id,
+        canonical_id,
+        rule_canonical,
+        absorbed_name,
+        absorbed_lat,
+        absorbed_lon
+    )
+VALUES
+    (?, ?, ?, ?, ?, ?);
+
+-- name: ListStopConsolidations :many
+SELECT
+    *
+FROM
+    stop_consolidations
+ORDER BY
+    absorbed_id ASC;
+
+-- name: ClearStopConsolidations :exec
+DELETE FROM stop_consolidations WHERE TRUE;

gtfsdb/schema.sql

@@ -596,3 +596,19 @@ CREATE INDEX IF NOT EXISTS idx_trips_time_window ON trips (max_departure_time, m
 
 -- migrate
 CREATE INDEX IF NOT EXISTS idx_stop_times_stop_revenue ON stop_times (stop_id, pickup_type, drop_off_type);
+
+-- migrate
+-- Stops removed by stop consolidation, one row per absorbed stop, rebuilt on every import.
+-- canonical_id is the stored stop that took over the absorbed stop's stop times, or NULL
+-- when they were dropped because the file's canonical stop is not stored. rule_canonical
+-- is the canonical stop as the consolidation file names it.
+CREATE TABLE
+    IF NOT EXISTS stop_consolidations (
+        absorbed_id TEXT PRIMARY KEY,
+        canonical_id TEXT,
+        rule_canonical TEXT NOT NULL,
+        absorbed_name TEXT,
+        absorbed_lat REAL,
+        absorbed_lon REAL,
+        FOREIGN KEY (canonical_id) REFERENCES stops (id)
+    ) STRICT;