feat(data-manager): wire live-traffic covered-way-id source + bootstrap way-to-edge map

The live-traffic writer needs /data/traffic/ways_to_edges.json, but getCoveredWayIds
was never wired into setupCron, so refreshWaysToEdges was always skipped and the map
never generated (writer failed ENOENT). Wire it to fetchCoveredWayIds (parsing way_ids
from the same OpenConditions /segments/speed.csv the writer consumes), and bootstrap the
map on startup when it is missing rather than only after a graph rebuild.

Author: Medformatik

services/data-manager/__tests__/cron.test.ts

@@ -344,6 +344,82 @@ describe("setupCron", () => {
       expect(fetchLiveTrafficCsv).not.toHaveBeenCalled();
       handles.stop();
     });
+
+    it("bootstraps the way-to-edge map on startup when it is missing", async () => {
+      const prev = process.env.DATA_DIR;
+      // The map path derives from DATA_DIR; a fresh tmp dir has no traffic/ subdir.
+      process.env.DATA_DIR = dataDir;
+      try {
+        const getCoveredWayIds = vi.fn().mockResolvedValue(new Set([123]));
+        const refreshWaysToEdges = vi.fn().mockResolvedValue({ wayCount: 1, edgeCount: 2 });
+        const handles = setupCron(
+          baseOptions({
+            openConditionsUrl: "http://openconditions-ingest:8080",
+            ensureTrafficExtract: vi.fn().mockResolvedValue({ built: false }),
+            getCoveredWayIds,
+            refreshWaysToEdges,
+          }),
+        );
+
+        await handles.runTrafficExtractStartupNow();
+
+        expect(getCoveredWayIds).toHaveBeenCalledTimes(1);
+        expect(refreshWaysToEdges).toHaveBeenCalledWith(new Set([123]), expect.anything());
+        handles.stop();
+      } finally {
+        if (prev === undefined) delete process.env.DATA_DIR;
+        else process.env.DATA_DIR = prev;
+      }
+    });
+
+    it("does not bootstrap the map on startup when it already exists", async () => {
+      const prev = process.env.DATA_DIR;
+      process.env.DATA_DIR = dataDir;
+      try {
+        mkdirSync(join(dataDir, "traffic"), { recursive: true });
+        writeFileSync(join(dataDir, "traffic", "ways_to_edges.json"), "{}");
+        const refreshWaysToEdges = vi.fn();
+        const handles = setupCron(
+          baseOptions({
+            openConditionsUrl: "http://openconditions-ingest:8080",
+            ensureTrafficExtract: vi.fn().mockResolvedValue({ built: false }),
+            getCoveredWayIds: vi.fn().mockResolvedValue(new Set([123])),
+            refreshWaysToEdges,
+          }),
+        );
+
+        await handles.runTrafficExtractStartupNow();
+
+        expect(refreshWaysToEdges).not.toHaveBeenCalled();
+        handles.stop();
+      } finally {
+        if (prev === undefined) delete process.env.DATA_DIR;
+        else process.env.DATA_DIR = prev;
+      }
+    });
+
+    it("skips the startup map bootstrap when no covered-way-id source is configured", async () => {
+      const prev = process.env.DATA_DIR;
+      process.env.DATA_DIR = dataDir;
+      try {
+        const refreshWaysToEdges = vi.fn();
+        const handles = setupCron(
+          baseOptions({
+            openConditionsUrl: "http://openconditions-ingest:8080",
+            ensureTrafficExtract: vi.fn().mockResolvedValue({ built: false }),
+            refreshWaysToEdges,
+          }),
+        );
+
+        await handles.runTrafficExtractStartupNow();
+
+        expect(refreshWaysToEdges).not.toHaveBeenCalled();
+        handles.stop();
+      } finally {
+        if (prev === undefined) delete process.env.DATA_DIR;
+        else process.env.DATA_DIR = prev;
+      }
+    });
   });
 });
 

services/data-manager/src/__tests__/covered-ways.test.ts

@@ -0,0 +1,32 @@
+import { describe, expect, it } from "vitest";
+import { parseCoveredWayIds } from "../jobs/traffic/covered-ways.js";
+
+describe("parseCoveredWayIds", () => {
+  const HEADER = "way_id,dir,current_kph,free_flow_kph,los";
+
+  it("extracts the way_id column, collapsing directions to unique way ids", () => {
+    const csv = [
+      HEADER,
+      "100118219,f,45,59.4,heavy",
+      "100118219,b,50,59.4,heavy",
+      "1021132485,f,101,75,free_flow",
+    ].join("\n");
+    expect(parseCoveredWayIds(csv)).toEqual(new Set([100118219, 1021132485]));
+  });
+
+  it("ignores the header, blank lines, and a trailing newline", () => {
+    const csv = `${HEADER}\n\n100118219,f,45,59.4,heavy\n\n1021132485,f,101,75,free_flow\n`;
+    expect(parseCoveredWayIds(csv)).toEqual(new Set([100118219, 1021132485]));
+  });
+
+  it("returns an empty set for a header-only feed", () => {
+    expect(parseCoveredWayIds(`${HEADER}\n`)).toEqual(new Set());
+  });
+
+  it("skips rows whose way_id is not an integer", () => {
+    const csv = [HEADER, ",f,45,59.4,heavy", "abc,f,45,59.4,heavy", "42,f,45,59.4,heavy"].join(
+      "\n",
+    );
+    expect(parseCoveredWayIds(csv)).toEqual(new Set([42]));
+  });
+});

services/data-manager/src/cron.ts

@@ -24,6 +24,7 @@ import {
   type RefreshWaysToEdgesResult,
   refreshWaysToEdges as refreshWaysToEdgesDefault,
   type WayEdge,
+  defaultOutputPath as waysToEdgesMapPath,
 } from "./jobs/traffic/ways-to-edges.js";
 import {
   type WriteLiveTrafficResult,
@@ -165,10 +166,11 @@ export interface CronSetupOptions {
   }) => Promise<boolean>;
   /**
    * Resolves the set of OSM way ids the live-speed writer currently covers.
-   * No default is wired: until the live-speed writer job supplies this (it
-   * already fetches the source feed, so it owns the covered-way-id list),
-   * the guard cron logs and skips the way-to-edge refresh rather than
-   * writing a map filtered down to nothing.
+   * The entrypoint wires this to `fetchCoveredWayIds` (the same OpenConditions
+   * speed feed the writer consumes) when OpenConditions is configured; when
+   * absent, the startup/guard way-to-edge refresh logs and skips rather than
+   * writing a map filtered down to nothing — the whole live-traffic chain then
+   * stays disabled.
    */
   getCoveredWayIds?: () => Promise<Set<number>>;
   /**
@@ -562,15 +564,6 @@ export function setupCron(options: CronSetupOptions): CronHandles {
   const ensureExtract = options.ensureTrafficExtract ?? ensureTrafficExtract;
   const checkExtractStale = options.isTrafficExtractStale ?? isTrafficExtractStale;
 
-  const runTrafficExtractStartup = async (): Promise<void> => {
-    try {
-      const result = await ensureExtract({ logger: log });
-      log.info("traffic-extract: startup check complete", { built: result.built });
-    } catch (err) {
-      log.error("traffic-extract: startup ensure failed", { err: (err as Error).message });
-    }
-  };
-
   const refreshWaysToEdges = options.refreshWaysToEdges ?? refreshWaysToEdgesDefault;
 
   // The way_id → GraphId map only changes when the graph rebuilds (a rebuild
@@ -593,6 +586,23 @@ export function setupCron(options: CronSetupOptions): CronHandles {
     }
   };
 
+  const runTrafficExtractStartup = async (): Promise<void> => {
+    try {
+      const result = await ensureExtract({ logger: log });
+      log.info("traffic-extract: startup check complete", { built: result.built });
+    } catch (err) {
+      log.error("traffic-extract: startup ensure failed", { err: (err as Error).message });
+    }
+    // A build (above, or the guard's rebuild) refreshes the way→edge map, but a
+    // traffic.tar that already exists from a prior boot leaves the map ungenerated
+    // on first run — the live writer then can't load it. Bootstrap it here when
+    // it's missing so the chain works without waiting for the next graph rebuild.
+    if (options.getCoveredWayIds && !existsSync(waysToEdgesMapPath())) {
+      log.info("ways-to-edges: map missing at startup, bootstrapping");
+      await runWaysToEdgesRefresh();
+    }
+  };
+
   const runTrafficExtractGuard = async (): Promise<void> => {
     try {
       const stale = await checkExtractStale({ logger: log });

services/data-manager/src/index.ts

@@ -17,6 +17,7 @@ import {
 import { type PoiSchedulerHandles, setupPoiIngestCron } from "./jobs/poi-ingest/scheduler.js";
 import { createPoiSingleFlight } from "./jobs/poi-ingest/single-flight.js";
 import { reconcileOrphanedJobs } from "./jobs/reconcile.js";
+import { fetchCoveredWayIds } from "./jobs/traffic/covered-ways.js";
 import { parseTransitousCountriesEnv } from "./jobs/transitous/internal.js";
 import { getSingleFlightController } from "./jobs/transitous/runtime.js";
 import { rootLogger } from "./logger.js";
@@ -138,13 +139,20 @@ app
     // fully ready before the first scheduled fire.
     const store = new StateStore(dataDir);
     const countries = parseTransitousCountriesEnv();
+    // The way→edge refresh restricts `valhalla_ways_to_edges` to the ways the
+    // live-traffic writer can update. Its covered-way-id source is the same
+    // OpenConditions speed feed the writer consumes, so it's only wired when
+    // OpenConditions is configured; otherwise the refresh (and the whole
+    // live-traffic chain) stays disabled.
+    const openConditionsUrl = process.env.OPENCONDITIONS_URL?.trim() ?? "";
     cronHandles = setupCron({
       dataDir,
       repoRoot,
       countries,
       store,
       singleFlight,
       logger: app.log,
+      getCoveredWayIds: openConditionsUrl ? () => fetchCoveredWayIds(openConditionsUrl) : undefined,
     });
 
     // Ensure the Valhalla traffic.tar extract exists before any job that

services/data-manager/src/jobs/traffic/covered-ways.ts

@@ -0,0 +1,45 @@
+import { fetchWithTimeout } from "../transitous/motis-probe.js";
+
+const COVERED_WAYS_FETCH_TIMEOUT_MS = 30_000;
+
+/**
+ * Parses the set of OSM way ids from an OpenConditions `/segments/speed.csv`
+ * body (header `way_id,dir,current_kph,free_flow_kph,los`; the `way_id` is the
+ * first column). Directions collapse to the underlying way id — the way→edge
+ * map `refreshWaysToEdges` builds is keyed by way id, and `valhalla_ways_to_edges`
+ * emits every directed edge for a way regardless of the sensor's direction.
+ */
+export function parseCoveredWayIds(csv: string): Set<number> {
+  const ids = new Set<number>();
+  const lines = csv.split("\n");
+  // Skip the header row; a trailing newline yields an empty final entry.
+  for (let i = 1; i < lines.length; i++) {
+    const line = lines[i]?.trim();
+    if (!line) continue;
+    const comma = line.indexOf(",");
+    const field = comma === -1 ? line : line.slice(0, comma);
+    // Number("") is 0, so an empty way_id column would slip through Number.isInteger.
+    if (field === "") continue;
+    const wayId = Number(field);
+    if (Number.isInteger(wayId)) ids.add(wayId);
+  }
+  return ids;
+}
+
+/**
+ * Resolves the covered-way-id list `refreshWaysToEdges` needs by fetching the
+ * same `${openConditionsUrl}/segments/speed.csv` feed the live-traffic writer
+ * consumes, then extracting its `way_id` column. Wired into `setupCron` as
+ * `getCoveredWayIds` so the way→edge refresh restricts `valhalla_ways_to_edges`
+ * to the ways the writer can actually update, instead of skipping the refresh.
+ */
+export async function fetchCoveredWayIds(openConditionsUrl: string): Promise<Set<number>> {
+  const res = await fetchWithTimeout(
+    `${openConditionsUrl}/segments/speed.csv`,
+    COVERED_WAYS_FETCH_TIMEOUT_MS,
+  );
+  if (!res.ok) {
+    throw new Error(`covered-ways: OpenConditions speed feed responded ${res.status}`);
+  }
+  return parseCoveredWayIds(await res.text());
+}

services/data-manager/src/jobs/traffic/ways-to-edges.ts

@@ -87,7 +87,8 @@ async function* defaultReadWayEdgesLines(container: string, path: string): Async
   }
 }
 
-function defaultOutputPath(): string {
+/** Path of the JSON way→edge map `refreshWaysToEdges` writes and `loadWaysToEdges` reads. */
+export function defaultOutputPath(): string {
   return join(process.env.DATA_DIR ?? "/data", "traffic", "ways_to_edges.json");
 }