Initial Watermaps import

scripts/download-geofabrik.sh

@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+OUT_DIR="${WATERMAPS_GEOFABRIK_DIR:-${SEA_COMPASS_GEOFABRIK_DIR:-data/geofabrik}}"
+GERMANY_BASE_URL="${WATERMAPS_GEOFABRIK_GERMANY_BASE_URL:-${SEA_COMPASS_GEOFABRIK_GERMANY_BASE_URL:-https://download.geofabrik.de/europe/germany}}"
+EUROPE_BASE_URL="${WATERMAPS_GEOFABRIK_EUROPE_BASE_URL:-${SEA_COMPASS_GEOFABRIK_EUROPE_BASE_URL:-https://download.geofabrik.de/europe}}"
+
+DEFAULT_REGIONS=(
+  germany
+  netherlands
+)
+
+regions=("$@")
+if [[ "${#regions[@]}" -eq 0 ]]; then
+  regions=("${DEFAULT_REGIONS[@]}")
+fi
+
+mkdir -p "$OUT_DIR"
+
+for region in "${regions[@]}"; do
+  file_name="${region}-latest.osm.pbf"
+  case "$region" in
+    germany|netherlands)
+      url="${EUROPE_BASE_URL}/${file_name}"
+      ;;
+    *)
+      url="${GERMANY_BASE_URL}/${file_name}"
+      ;;
+  esac
+  target="${OUT_DIR}/${file_name}"
+  checksum_target="${target}.md5"
+
+  echo "Downloading $url.md5"
+  curl --fail --location --retry 5 --retry-delay 5 --output "$checksum_target" "${url}.md5"
+
+  expected_checksum="$(awk 'NR == 1 { print $1 }' "$checksum_target")"
+  if [[ -f "$target" ]] && [[ "$(md5sum "$target" | awk '{ print $1 }')" == "$expected_checksum" ]]; then
+    echo "$(basename "$target"): already current"
+    continue
+  fi
+
+  echo "Downloading $url"
+  curl --fail --location --continue-at - --retry 5 --retry-delay 5 --output "$target" "$url"
+
+  actual_checksum="$(md5sum "$target" | awk '{ print $1 }')"
+  if [[ -z "$expected_checksum" || "$actual_checksum" != "$expected_checksum" ]]; then
+    echo "Checksum verification failed for $target" >&2
+    exit 1
+  fi
+  echo "$(basename "$target"): OK"
+done

scripts/import-geofabrik-docker.sh

@@ -0,0 +1,32 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+IMAGE_NAME="${WATERMAPS_IMPORT_IMAGE:-${SEA_COMPASS_IMPORT_IMAGE:-watermaps-geofabrik-importer}}"
+DATABASE_URL="${DATABASE_URL:-postgres://seacompass:seacompass@localhost:5432/seacompass}"
+
+if [[ "$#" -lt 1 ]]; then
+  echo "Usage: DATABASE_URL=postgres://... ./scripts/import-geofabrik-docker.sh data/geofabrik/region-latest.osm.pbf [...]" >&2
+  exit 1
+fi
+
+docker build -f "$ROOT_DIR/docker/geofabrik-import/Dockerfile" -t "$IMAGE_NAME" "$ROOT_DIR"
+
+for input_path in "$@"; do
+  host_path="$(realpath "$input_path")"
+  case "$host_path" in
+    "$ROOT_DIR"/*) ;;
+    *)
+      echo "PBF file must be inside the project directory so it can be mounted into the importer: $input_path" >&2
+      exit 1
+      ;;
+  esac
+
+  relative_path="$(realpath --relative-to="$ROOT_DIR" "$host_path")"
+  docker run --rm \
+    --network host \
+    -v "$ROOT_DIR:/workspace" \
+    -e DATABASE_URL="$DATABASE_URL" \
+    "$IMAGE_NAME" \
+    /workspace/scripts/import-geofabrik.sh "/workspace/$relative_path"
+done

scripts/import-geofabrik.sh

@@ -0,0 +1,67 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if ! command -v osmium >/dev/null 2>&1; then
+  echo "osmium is required for filtering OSM PBF files." >&2
+  exit 1
+fi
+
+if ! command -v psql >/dev/null 2>&1; then
+  echo "psql is required for preparing the PostGIS schema." >&2
+  exit 1
+fi
+
+if ! command -v node >/dev/null 2>&1; then
+  echo "node is required for loading filtered OSM features into PostGIS." >&2
+  exit 1
+fi
+
+PBF_PATH="${1:-}"
+DATABASE_URL="${DATABASE_URL:-postgres://seacompass:seacompass@localhost:55432/seacompass}"
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+
+if [[ -z "$PBF_PATH" || ! -f "$PBF_PATH" ]]; then
+  echo "Usage: DATABASE_URL=postgres://... ./scripts/import-geofabrik.sh /path/to/europe-latest.osm.pbf" >&2
+  exit 1
+fi
+
+WORKDIR="$(mktemp -d)"
+trap 'rm -rf "$WORKDIR"' EXIT
+
+osmium tags-filter "$PBF_PATH" \
+  nwr/seamark:type \
+  nwr/lock=yes \
+  nwr/water=lock \
+  nwr/waterway=lock \
+  nwr/waterway=lock_gate \
+  nwr/obstacle=lock \
+  nwr/leisure=marina \
+  nwr/harbour \
+  nwr/industrial=port \
+  nwr/landuse=harbour \
+  nwr/landuse=port \
+  nwr/waterway=dock \
+  w/waterway=fairway \
+  w/waterway=canal \
+  w/waterway=river \
+  w/route=ferry \
+  w/natural=water \
+  w/water=lake \
+  w/boat=yes \
+  w/ship=yes \
+  nwr/bridge \
+  r/natural=water \
+  r/water=lake \
+  r/route=ferry \
+  -o "$WORKDIR/marine.osm.pbf" \
+  --overwrite
+
+osmium export "$WORKDIR/marine.osm.pbf" \
+  --format geojsonseq \
+  --add-unique-id type_id \
+  --attributes type,id \
+  -o "$WORKDIR/marine.geojsonseq" \
+  --overwrite
+
+psql "$DATABASE_URL" -f "$ROOT_DIR/database/schema.sql"
+DATABASE_URL="$DATABASE_URL" WATERMAPS_IMPORT_SOURCE="$(basename "$PBF_PATH" .osm.pbf)" node "$ROOT_DIR/scripts/load-osm-fairways.mjs" "$WORKDIR/marine.geojsonseq"

scripts/load-osm-fairways.mjs

@@ -0,0 +1,383 @@
+#!/usr/bin/env node
+import { createReadStream } from "node:fs";
+import { createInterface } from "node:readline";
+import pg from "pg";
+import {
+  featureName,
+  isLockFeature,
+  normalizedValue,
+  sourceId
+} from "./osm-marine-classification.mjs";
+
+const inputPath = process.argv[2];
+const databaseUrl = process.env.DATABASE_URL ?? "postgres://seacompass:seacompass@localhost:55432/seacompass";
+const importSource =
+  process.env.WATERMAPS_IMPORT_SOURCE ??
+  process.env.SEA_COMPASS_IMPORT_SOURCE ??
+  "unknown-pbf";
+
+if (!inputPath) {
+  console.error("Usage: DATABASE_URL=postgres://... node scripts/load-osm-fairways.mjs /path/to/marine.geojsonseq");
+  process.exit(1);
+}
+
+const { Client } = pg;
+const client = new Client({ connectionString: databaseUrl });
+
+const FEATURE_BATCH_SIZE = 500;
+const EDGE_BATCH_SIZE = 500;
+const featureBatch = [];
+const edgeBatch = [];
+let seenFeatures = 0;
+let importedFeatures = 0;
+let importedEdges = 0;
+
+const truthy = new Set(["yes", "true", "1", "designated", "permissive"]);
+const falsy = new Set(["no", "false", "0", "private", "prohibited", "restricted"]);
+const harbourSeamarkTypes = new Set(["harbour", "harbour_basin", "marina"]);
+const inactiveWaterwayValues = new Set(["abandoned", "construction", "disused", "proposed"]);
+
+function getString(value) {
+  return typeof value === "string" ? value.trim() : "";
+}
+
+function hasKey(properties, key) {
+  return Object.prototype.hasOwnProperty.call(properties, key) && properties[key] != null;
+}
+
+function hasActiveValue(value) {
+  const normalized = normalizedValue(value);
+  return Boolean(normalized) && !["no", "false", "0"].includes(normalized);
+}
+
+function isHarbourFeature(properties) {
+  const seamarkType = normalizedValue(properties["seamark:type"]);
+  const waterway = normalizedValue(properties.waterway);
+
+  return (
+    harbourSeamarkTypes.has(seamarkType) ||
+    normalizedValue(properties.leisure) === "marina" ||
+    hasActiveValue(properties.harbour) ||
+    normalizedValue(properties.industrial) === "port" ||
+    ["harbour", "port"].includes(normalizedValue(properties.landuse)) ||
+    waterway === "dock"
+  );
+}
+
+function layerFor(properties) {
+  const seamarkType = normalizedValue(properties["seamark:type"]);
+  const waterway = normalizedValue(properties.waterway);
+  const route = normalizedValue(properties.route);
+
+  if (isLockFeature(properties)) {
+    return "locks";
+  }
+
+  if (isHarbourFeature(properties)) {
+    return "harbours";
+  }
+
+  if (hasKey(properties, "bridge")) {
+    return "bridges";
+  }
+
+  if (["fairway", "navigation_line", "recommended_track"].includes(seamarkType)) {
+    return "fairways";
+  }
+
+  if (["fairway", "canal", "river"].includes(waterway) || route === "ferry") {
+    return "fairways";
+  }
+
+  if (seamarkType) {
+    return "seamarks";
+  }
+
+  return "waterways";
+}
+
+function isRelevantFeature(properties) {
+  const waterway = normalizedValue(properties.waterway);
+
+  return (
+    isLockFeature(properties) ||
+    isHarbourFeature(properties) ||
+    hasKey(properties, "seamark:type") ||
+    ["fairway", "canal", "river", "dock"].includes(waterway) ||
+    normalizedValue(properties.route) === "ferry" ||
+    normalizedValue(properties.natural) === "water" ||
+    normalizedValue(properties.water) === "lake" ||
+    isTruthy(properties.boat) ||
+    isTruthy(properties.ship) ||
+    hasKey(properties, "bridge")
+  );
+}
+
+function parseDepth(properties) {
+  const raw = [
+    properties["seamark:fairway:minimum_depth"],
+    properties["seamark:recommended_track:minimum_depth"],
+    properties["seamark:navigation_line:minimum_depth"],
+    properties["depth"],
+    properties["min_depth"]
+  ]
+    .map(getString)
+    .find(Boolean);
+
+  if (!raw) {
+    return null;
+  }
+
+  const normalized = raw.replace(",", ".");
+  const match = normalized.match(/-?\d+(?:\.\d+)?/);
+  if (!match) {
+    return null;
+  }
+
+  const value = Number(match[0]);
+  return Number.isFinite(value) ? value : null;
+}
+
+function isTruthy(value) {
+  return truthy.has(getString(value).toLowerCase());
+}
+
+function isFalsy(value) {
+  return falsy.has(normalizedValue(value));
+}
+
+function isInactiveOrUnderConstruction(properties) {
+  const waterway = normalizedValue(properties.waterway);
+  if (inactiveWaterwayValues.has(waterway)) {
+    return true;
+  }
+
+  const lifecycleKeys = [
+    "abandoned",
+    "construction",
+    "disused",
+    "proposed",
+    "abandoned:waterway",
+    "construction:waterway",
+    "disused:waterway",
+    "proposed:waterway",
+    "disused:route"
+  ];
+  return lifecycleKeys.some((key) => hasActiveValue(properties[key]));
+}
+
+function isRoutable(properties, geometry) {
+  if (!geometry || !["LineString", "MultiLineString"].includes(geometry.type)) {
+    return false;
+  }
+
+  const seamarkType = normalizedValue(properties["seamark:type"]);
+  const waterway = normalizedValue(properties.waterway);
+  const route = normalizedValue(properties.route);
+
+  if (
+    ["access", "boat", "ship", "motorboat"].some((key) => isFalsy(properties[key])) ||
+    isInactiveOrUnderConstruction(properties) ||
+    normalizedValue(properties.tunnel) === "culvert"
+  ) {
+    return false;
+  }
+
+  return (
+    ["navigation_line", "recommended_track", "fairway"].includes(seamarkType) ||
+    ["fairway", "canal", "river"].includes(waterway) ||
+    route === "ferry" ||
+    isTruthy(properties.boat) ||
+    isTruthy(properties.ship)
+  );
+}
+
+function isClosedLine(coords) {
+  if (!Array.isArray(coords) || coords.length < 2) {
+    return false;
+  }
+
+  const first = coords[0];
+  const last = coords[coords.length - 1];
+  return Array.isArray(first) && Array.isArray(last) && first[0] === last[0] && first[1] === last[1];
+}
+
+function edgeGeometries(geometry) {
+  if (!geometry) {
+    return [];
+  }
+
+  if (geometry.type === "LineString") {
+    return isClosedLine(geometry.coordinates)
+      ? []
+      : [{ type: "LineString", coordinates: geometry.coordinates }];
+  }
+
+  if (geometry.type === "MultiLineString") {
+    return geometry.coordinates
+      .filter((line) => Array.isArray(line) && line.length >= 2 && !isClosedLine(line))
+      .map((line) => ({ type: "LineString", coordinates: line }));
+  }
+
+  return [];
+}
+
+function pushFeature(record) {
+  featureBatch.push(record);
+}
+
+function pushEdge(record) {
+  edgeBatch.push(record);
+}
+
+async function flushFeatures() {
+  if (!featureBatch.length) {
+    return;
+  }
+
+  const uniqueFeatures = [
+    ...new Map(featureBatch.map((feature) => [`${feature.layer}:${feature.sourceId}`, feature])).values()
+  ];
+  const params = [];
+  const values = uniqueFeatures.map((feature, index) => {
+    const offset = index * 5;
+    params.push(
+      feature.layer,
+      feature.sourceId,
+      feature.name,
+      JSON.stringify(feature.properties),
+      JSON.stringify(feature.geometry)
+    );
+    return `($${offset + 1}, 'osm', $${offset + 2}, $${offset + 3}, $${offset + 4}::jsonb, ST_SetSRID(ST_GeomFromGeoJSON($${offset + 5}), 4326))`;
+  });
+
+  await client.query(`
+    WITH incoming (layer, source, source_id, name, properties, geom) AS (
+      VALUES ${values.join(",")}
+    ),
+    removed_stale_classifications AS (
+      DELETE FROM marine_features existing
+      USING incoming
+      WHERE existing.source = incoming.source
+        AND existing.source_id = incoming.source_id
+        AND existing.layer <> incoming.layer
+    )
+    INSERT INTO marine_features (layer, source, source_id, name, properties, geom)
+    SELECT layer, source, source_id, name, properties, geom
+    FROM incoming
+    ON CONFLICT (source, source_id, layer) WHERE source_id IS NOT NULL
+    DO UPDATE SET
+      name = EXCLUDED.name,
+      properties = EXCLUDED.properties,
+      geom = EXCLUDED.geom,
+      updated_at = now()
+  `, params);
+
+  importedFeatures += uniqueFeatures.length;
+  featureBatch.length = 0;
+}
+
+async function flushEdges() {
+  if (!edgeBatch.length) {
+    return;
+  }
+
+  const uniqueEdges = [...new Map(edgeBatch.map((edge) => [edge.sourceId, edge])).values()];
+  const params = [];
+  const values = uniqueEdges.map((edge, index) => {
+    const offset = index * 5;
+    params.push(
+      edge.sourceId,
+      edge.name,
+      edge.depth,
+      JSON.stringify(edge.properties),
+      JSON.stringify(edge.geometry)
+    );
+    return `('osm', $${offset + 1}, $${offset + 2}, $${offset + 3}, $${offset + 4}::jsonb, ST_SetSRID(ST_GeomFromGeoJSON($${offset + 5}), 4326)::geometry(LineString, 4326))`;
+  });
+
+  await client.query(`
+    INSERT INTO marine_fairway_edges (source, source_id, name, min_depth_m, properties, geom)
+    VALUES ${values.join(",")}
+    ON CONFLICT (source, source_id) WHERE source_id IS NOT NULL
+    DO UPDATE SET
+      name = EXCLUDED.name,
+      min_depth_m = EXCLUDED.min_depth_m,
+      properties = EXCLUDED.properties,
+      geom = EXCLUDED.geom,
+      updated_at = now()
+  `, params);
+
+  importedEdges += uniqueEdges.length;
+  edgeBatch.length = 0;
+}
+
+async function flushAll() {
+  await flushFeatures();
+  await flushEdges();
+}
+
+await client.connect();
+
+try {
+  const lines = createInterface({
+    input: createReadStream(inputPath, { encoding: "utf8" }),
+    crlfDelay: Infinity
+  });
+
+  for await (const line of lines) {
+    let trimmed = line.trim();
+    if (trimmed.charCodeAt(0) === 0x1e) {
+      trimmed = trimmed.slice(1).trimStart();
+    }
+
+    if (!trimmed) {
+      continue;
+    }
+
+    seenFeatures += 1;
+    const feature = JSON.parse(trimmed);
+    const properties = feature.properties ?? {};
+    const geometry = feature.geometry;
+    if (!isRelevantFeature(properties)) {
+      continue;
+    }
+
+    const id = sourceId(feature, seenFeatures, importSource);
+    const name = featureName(properties);
+
+    if (geometry) {
+      pushFeature({
+        layer: layerFor(properties),
+        sourceId: id,
+        name,
+        properties,
+        geometry
+      });
+    }
+
+    if (isRoutable(properties, geometry)) {
+      const linestrings = edgeGeometries(geometry);
+      const depth = parseDepth(properties);
+      linestrings.forEach((edgeGeometry, index) => {
+        pushEdge({
+          sourceId: `${id}/line/${index}`,
+          name,
+          depth,
+          properties,
+          geometry: edgeGeometry
+        });
+      });
+    }
+
+    if (featureBatch.length >= FEATURE_BATCH_SIZE || edgeBatch.length >= EDGE_BATCH_SIZE) {
+      await flushAll();
+    }
+  }
+
+  await flushAll();
+  console.log(`Read ${seenFeatures} OSM marine features.`);
+  console.log(`Queued ${importedFeatures} marine feature rows and ${importedEdges} routable fairway edge rows.`);
+} finally {
+  await client.end();
+}

scripts/osm-marine-classification.mjs

@@ -0,0 +1,54 @@
+const truthy = new Set(["yes", "true", "1", "designated", "permissive"]);
+const lockSeamarkTypes = new Set(["lock", "lock_basin", "lock_gate"]);
+
+function stringValue(value) {
+  return typeof value === "string" ? value.trim() : "";
+}
+
+export function normalizedValue(value) {
+  return stringValue(value).toLowerCase();
+}
+
+export function isLockFeature(properties) {
+  const seamarkType = normalizedValue(properties["seamark:type"]);
+  const seamarkGateCategory = normalizedValue(properties["seamark:gate:category"]);
+  const waterway = normalizedValue(properties.waterway);
+
+  return (
+    truthy.has(normalizedValue(properties.lock)) ||
+    ["lock", "lock_gate"].includes(waterway) ||
+    normalizedValue(properties.water) === "lock" ||
+    lockSeamarkTypes.has(seamarkType) ||
+    (seamarkType === "gate" && ["lock", "lock_gate"].includes(seamarkGateCategory)) ||
+    normalizedValue(properties.obstacle) === "lock"
+  );
+}
+
+export function sourceId(feature, fallbackIndex, importSource) {
+  const properties = feature.properties ?? {};
+  const osmId = properties["@id"] ?? properties.osm_id ?? properties.osmId;
+  const osmType = normalizedValue(properties["@type"] ?? properties.osm_type ?? properties.osmType);
+  const osmTypePrefix = { node: "n", way: "w", relation: "r" }[osmType];
+  const normalizedOsmId = String(osmId ?? "").match(/\d+/)?.[0];
+  if (osmTypePrefix && normalizedOsmId) {
+    return `${osmTypePrefix}${normalizedOsmId}`;
+  }
+
+  const rawId = feature.id ?? properties.id ?? fallbackIndex;
+  const rawType = properties.type ?? "osm";
+  const id = String(rawId);
+  if (id.includes("/") || /^[nwr]\d+$/.test(id)) {
+    return id;
+  }
+
+  return `${rawType}/${importSource}/${id}`;
+}
+
+export function featureName(properties) {
+  return (
+    stringValue(properties.lock_name) ||
+    stringValue(properties.name) ||
+    stringValue(properties["seamark:name"]) ||
+    null
+  );
+}

scripts/sync-euris-locks.mjs

@@ -0,0 +1,547 @@
+#!/usr/bin/env node
+
+import { pathToFileURL } from "node:url";
+
+export const EURIS_BASE_URL = "https://www.eurisportal.eu";
+export const EURIS_COMPACT_LOCKS_URL = `${EURIS_BASE_URL}/visuris/api/Locks_v2/GetCompactLocks`;
+export const EURIS_RIS_INDEX_URL = `${EURIS_BASE_URL}/visuris/api/RisIndices_v2/GetRISIndexObjects`;
+export const EURIS_LOCK_DETAIL_URL = `${EURIS_BASE_URL}/visuris/api/Locks_v2/GetLock`;
+export const MAX_PAGE_SIZE = 100;
+export const MAX_DETAIL_LIMIT = 20;
+export const DETAIL_REQUEST_DELAY_MS = 1_000;
+
+const DEFAULT_TIMEOUT_MS = 15_000;
+const DEFAULT_MAX_RETRIES = 4;
+const MAX_PAGES = 100;
+
+const sleep = (milliseconds) =>
+  new Promise((resolve) => setTimeout(resolve, milliseconds));
+
+export class EurisRequestError extends Error {
+  constructor(message, { status, url, cause } = {}) {
+    super(message, { cause });
+    this.name = "EurisRequestError";
+    this.status = status;
+    this.url = url;
+  }
+}
+
+function nonEmptyString(value) {
+  if (value === null || value === undefined) return null;
+  const normalized = String(value).trim();
+  return normalized || null;
+}
+
+function firstNonEmpty(...values) {
+  for (const value of values) {
+    const normalized = nonEmptyString(value);
+    if (normalized) return normalized;
+  }
+  return null;
+}
+
+function finiteCoordinate(value, minimum, maximum) {
+  const coordinate = Number(value);
+  return Number.isFinite(coordinate) && coordinate >= minimum && coordinate <= maximum
+    ? coordinate
+    : null;
+}
+
+function unique(values) {
+  return [...new Set(values.filter(Boolean))];
+}
+
+export function parseCountries(value = "DE") {
+  const countries = unique(
+    String(value || "DE")
+      .split(/[;,\s]+/u)
+      .map((country) => country.trim().toUpperCase())
+      .filter(Boolean),
+  );
+
+  if (countries.length === 0) return ["DE"];
+  const invalid = countries.find((country) => !/^[A-Z]{2}$/u.test(country));
+  if (invalid) {
+    throw new Error(`Ungültiger ISO-Ländercode in EURIS_COUNTRIES: ${invalid}`);
+  }
+  return countries;
+}
+
+export function parseDetailLimit(value) {
+  if (value === undefined || value === null || value === "") return 0;
+  const parsed = Number.parseInt(String(value), 10);
+  if (!Number.isFinite(parsed) || parsed < 0) {
+    throw new Error("EURIS_DETAIL_LIMIT muss eine nichtnegative Ganzzahl sein.");
+  }
+  return Math.min(parsed, MAX_DETAIL_LIMIT);
+}
+
+export function parseBoolean(value) {
+  return /^(1|true|yes|on)$/iu.test(String(value || "").trim());
+}
+
+export function normalizePhones(value) {
+  if (!value) return [];
+
+  const values = Array.isArray(value) ? value : [value];
+  return unique(
+    values
+      .flatMap((entry) => String(entry).split(/\s*[;|]\s*/u))
+      .map((phone) => phone.trim().replace(/\s+/gu, " "))
+      .map((phone) => phone.replace(/^00(?=\d)/u, "+"))
+      .filter(Boolean),
+  );
+}
+
+export function normalizeVhf(value) {
+  return nonEmptyString(value)?.replace(/\s+/gu, " ") ?? null;
+}
+
+export function retryAfterMilliseconds(value, now = Date.now()) {
+  const normalized = nonEmptyString(value);
+  if (!normalized) return null;
+
+  const seconds = Number(normalized);
+  if (Number.isFinite(seconds) && seconds >= 0) return Math.ceil(seconds * 1_000);
+
+  const date = Date.parse(normalized);
+  if (!Number.isFinite(date)) return null;
+  return Math.max(0, date - now);
+}
+
+function backoffMilliseconds(attempt) {
+  return Math.min(30_000, 1_000 * 2 ** attempt);
+}
+
+async function responseBodySnippet(response) {
+  try {
+    return (await response.text()).replace(/\s+/gu, " ").slice(0, 300);
+  } catch {
+    return "";
+  }
+}
+
+export async function requestJson(
+  url,
+  {
+    fetchImpl = globalThis.fetch,
+    token,
+    timeoutMs = DEFAULT_TIMEOUT_MS,
+    maxRetries = DEFAULT_MAX_RETRIES,
+    sleepImpl = sleep,
+    now = Date.now,
+  } = {},
+) {
+  if (typeof fetchImpl !== "function") throw new Error("Keine Fetch-Implementierung verfügbar.");
+
+  for (let attempt = 0; attempt <= maxRetries; attempt += 1) {
+    const controller = new AbortController();
+    const timeout = setTimeout(
+      () => controller.abort(new Error(`EuRIS-Anfrage nach ${timeoutMs} ms abgebrochen.`)),
+      timeoutMs,
+    );
+
+    try {
+      const headers = new Headers({ Accept: "application/json" });
+      if (token) headers.set("Authorization", `Bearer ${token}`);
+
+      const response = await fetchImpl(url, {
+        method: "GET",
+        headers,
+        signal: controller.signal,
+      });
+
+      if (response.ok) {
+        const body = await response.text();
+        return body ? JSON.parse(body) : null;
+      }
+
+      const retryable = [429, 502, 503].includes(response.status);
+      if (!retryable || attempt === maxRetries) {
+        const snippet = await responseBodySnippet(response);
+        throw new EurisRequestError(
+          `EuRIS antwortete mit HTTP ${response.status}${snippet ? `: ${snippet}` : ""}`,
+          { status: response.status, url },
+        );
+      }
+
+      const retryAfter = retryAfterMilliseconds(response.headers.get("retry-after"), now());
+      await sleepImpl(retryAfter ?? backoffMilliseconds(attempt));
+    } catch (error) {
+      if (error instanceof EurisRequestError) throw error;
+      if (attempt === maxRetries) {
+        throw new EurisRequestError(`EuRIS-Anfrage fehlgeschlagen: ${error.message}`, {
+          url,
+          cause: error,
+        });
+      }
+      await sleepImpl(backoffMilliseconds(attempt));
+    } finally {
+      clearTimeout(timeout);
+    }
+  }
+
+  throw new EurisRequestError("EuRIS-Anfrage ohne Ergebnis beendet.", { url });
+}
+
+export function compactLocksFilter(countries) {
+  return countries.map((country) => `startswith(locode,'${country}')`).join(" or ");
+}
+
+export function risIndexFilter(countries) {
+  const countryFilter = countries.map((country) => `countryCode eq '${country}'`).join(" or ");
+  return `(${countryFilter}) and function eq 'lokare'`;
+}
+
+export async function fetchODataPages({
+  endpoint,
+  filter,
+  orderBy,
+  idForItem,
+  pageSize = MAX_PAGE_SIZE,
+  requestOptions,
+}) {
+  const top = Math.max(1, Math.min(MAX_PAGE_SIZE, Number(pageSize) || MAX_PAGE_SIZE));
+  const records = new Map();
+  let skip = 0;
+  let expectedCount = null;
+
+  for (let page = 0; page < MAX_PAGES; page += 1) {
+    const url = new URL(endpoint);
+    url.searchParams.set("$filter", filter);
+    url.searchParams.set("$orderby", orderBy);
+    url.searchParams.set("$top", String(top));
+    url.searchParams.set("$skip", String(skip));
+    url.searchParams.set("$count", "true");
+
+    const payload = await requestJson(url, requestOptions);
+    const items = Array.isArray(payload?.items) ? payload.items : [];
+    if (expectedCount === null && Number.isFinite(Number(payload?.count))) {
+      expectedCount = Number(payload.count);
+    }
+
+    for (const item of items) {
+      const identifier = nonEmptyString(idForItem(item))?.toUpperCase();
+      if (identifier) records.set(identifier, item);
+    }
+
+    skip += items.length;
+    if (items.length < top || items.length === 0 || (expectedCount !== null && skip >= expectedCount)) {
+      return { items: [...records.values()], expectedCount, requestedCount: skip };
+    }
+  }
+
+  throw new Error(`EuRIS-Paginierung überschritt ${MAX_PAGES} Seiten für ${endpoint}.`);
+}
+
+function splitContactValues(values) {
+  if (!values) return [];
+  const entries = Array.isArray(values) ? values : [values];
+  return unique(
+    entries
+      .flatMap((entry) => String(entry).split(/\s*[;,|]\s*/u))
+      .map((entry) => entry.trim())
+      .filter(Boolean),
+  );
+}
+
+export function extractDetailFields(detail) {
+  if (!detail || typeof detail !== "object") return {};
+
+  const facility = detail.facility && typeof detail.facility === "object" ? detail.facility : {};
+  const contacts = Array.isArray(facility.contacts) ? facility.contacts : [];
+  const phones = normalizePhones(contacts.flatMap((contact) => contact?.phones || []));
+  const emails = splitContactValues(contacts.flatMap((contact) => contact?.emails || []));
+  const websites = splitContactValues(contacts.flatMap((contact) => contact?.urls || []));
+  const addressParts = [facility.street, facility.postCode, facility.city, facility.country]
+    .map(nonEmptyString)
+    .filter(Boolean);
+
+  return {
+    operator: firstNonEmpty(
+      facility.operator,
+      facility.owner,
+      ...contacts.map((contact) => contact?.company),
+      detail.waterwayAuthority,
+    ),
+    phones,
+    email: emails[0] ?? null,
+    website: websites[0] ?? null,
+    address: addressParts.length > 0 ? addressParts.join(", ") : null,
+  };
+}
+
+export function buildLockRecord({ compact, ris, detail, fetchedAt }) {
+  const sourceId = firstNonEmpty(ris?.isrs, compact?.locode)?.toUpperCase();
+  if (!sourceId) return null;
+
+  const longitude = finiteCoordinate(ris?.lon ?? ris?.longitude, -180, 180);
+  const latitude = finiteCoordinate(ris?.lat ?? ris?.latitude, -90, 90);
+  if (longitude === null || latitude === null) return null;
+
+  const detailFields = extractDetailFields(detail);
+  const compactPhones = normalizePhones(compact?.contactPhone);
+  const phones = unique([...compactPhones, ...(detailFields.phones || [])]);
+  const phone = phones[0] ?? null;
+  const name = firstNonEmpty(
+    compact?.nationalObjectName,
+    compact?.objectName,
+    ris?.nationalObjectName,
+    ris?.objectName,
+    sourceId,
+  );
+  const waterway = firstNonEmpty(compact?.waterwayName, ris?.fairwaySectionName, ris?.fairwayName);
+  const detailUrl = new URL(EURIS_LOCK_DETAIL_URL);
+  detailUrl.searchParams.set("isrs", sourceId);
+
+  const properties = {
+    isrs: sourceId,
+    "ref:EU:RIS": sourceId,
+    name,
+    lock_name: name,
+    country: firstNonEmpty(ris?.countryCode, sourceId.slice(0, 2)),
+    waterway_name: waterway,
+    waterwayName: waterway,
+    hectom: compact?.hectom ?? ris?.hectom ?? null,
+    phone,
+    phones,
+    phone_raw: nonEmptyString(compact?.contactPhone),
+    "contact:phone": phone,
+    vhf: normalizeVhf(compact?.comcha),
+    operator: detailFields.operator ?? null,
+    email: detailFields.email ?? null,
+    website: detailFields.website ?? null,
+    address: detailFields.address ?? null,
+    upstream_source: nonEmptyString(ris?.source),
+    upstreamSource: nonEmptyString(ris?.source),
+    data_source: nonEmptyString(ris?.source),
+    source_url: detailUrl.toString(),
+    compact_source_url: EURIS_COMPACT_LOCKS_URL,
+    ris_source_url: EURIS_RIS_INDEX_URL,
+    fetched_at: fetchedAt,
+    fetchedAt,
+  };
+
+  return { sourceId, name, longitude, latitude, properties };
+}
+
+async function fetchLockDetail(sourceId, requestOptions) {
+  const url = new URL(EURIS_LOCK_DETAIL_URL);
+  url.searchParams.set("isrs", sourceId);
+  return requestJson(url, requestOptions);
+}
+
+export async function collectEurisLocks({
+  countries = ["DE"],
+  detailLimit = 0,
+  pageSize = MAX_PAGE_SIZE,
+  fetchImpl = globalThis.fetch,
+  token,
+  timeoutMs = DEFAULT_TIMEOUT_MS,
+  maxRetries = DEFAULT_MAX_RETRIES,
+  sleepImpl = sleep,
+  fetchedAt = new Date().toISOString(),
+  logger = console,
+} = {}) {
+  const normalizedCountries = parseCountries(
+    Array.isArray(countries) ? countries.join(",") : countries,
+  );
+  const requestOptions = { fetchImpl, token, timeoutMs, maxRetries, sleepImpl };
+
+  const compactPage = await fetchODataPages({
+    endpoint: EURIS_COMPACT_LOCKS_URL,
+    filter: compactLocksFilter(normalizedCountries),
+    orderBy: "objectName,waterwayName,hectom",
+    idForItem: (item) => item.locode,
+    pageSize,
+    requestOptions,
+  });
+  const risPage = await fetchODataPages({
+    endpoint: EURIS_RIS_INDEX_URL,
+    filter: risIndexFilter(normalizedCountries),
+    orderBy: "objectName",
+    idForItem: (item) => item.isrs,
+    pageSize,
+    requestOptions,
+  });
+
+  const compactByIsrs = new Map(
+    compactPage.items
+      .map((item) => [nonEmptyString(item.locode)?.toUpperCase(), item])
+      .filter(([identifier]) => identifier),
+  );
+  const risByIsrs = new Map(
+    risPage.items
+      .map((item) => [nonEmptyString(item.isrs)?.toUpperCase(), item])
+      .filter(([identifier]) => identifier),
+  );
+  const joinedIds = [...risByIsrs.keys()].filter((sourceId) => compactByIsrs.has(sourceId)).sort();
+  const details = new Map();
+  let detailFailures = 0;
+  const effectiveDetailLimit = Math.min(parseDetailLimit(detailLimit), joinedIds.length);
+
+  for (let index = 0; index < effectiveDetailLimit; index += 1) {
+    const sourceId = joinedIds[index];
+    try {
+      details.set(sourceId, await fetchLockDetail(sourceId, requestOptions));
+    } catch (error) {
+      detailFailures += 1;
+      logger.warn?.(`EuRIS-Details für ${sourceId} konnten nicht geladen werden: ${error.message}`);
+    }
+    if (index + 1 < effectiveDetailLimit) await sleepImpl(DETAIL_REQUEST_DELAY_MS);
+  }
+
+  const records = [];
+  let skippedWithoutCoordinates = 0;
+  for (const [sourceId, ris] of [...risByIsrs.entries()].sort(([left], [right]) => left.localeCompare(right))) {
+    const record = buildLockRecord({
+      compact: compactByIsrs.get(sourceId),
+      ris,
+      detail: details.get(sourceId),
+      fetchedAt,
+    });
+    if (record) records.push(record);
+    else skippedWithoutCoordinates += 1;
+  }
+
+  return {
+    records,
+    stats: {
+      countries: normalizedCountries,
+      compactLocks: compactByIsrs.size,
+      risLocks: risByIsrs.size,
+      joinedLocks: joinedIds.length,
+      risOnlyLocks: [...risByIsrs.keys()].filter((sourceId) => !compactByIsrs.has(sourceId)).length,
+      compactOnlyLocks: [...compactByIsrs.keys()].filter((sourceId) => !risByIsrs.has(sourceId)).length,
+      storedLocks: records.length,
+      skippedWithoutCoordinates,
+      detailsRequested: effectiveDetailLimit,
+      detailFailures,
+      fetchedAt,
+    },
+  };
+}
+
+export async function writeEurisLocks(records, { databaseUrl = process.env.DATABASE_URL } = {}) {
+  if (!databaseUrl) throw new Error("DATABASE_URL ist für den EuRIS-Import erforderlich.");
+  if (records.length === 0) return 0;
+
+  const { default: pg } = await import("pg");
+  const client = new pg.Client({ connectionString: databaseUrl });
+  await client.connect();
+
+  try {
+    await client.query("BEGIN");
+    const batchSize = 100;
+    for (let offset = 0; offset < records.length; offset += batchSize) {
+      const batch = records.slice(offset, offset + batchSize);
+      const values = [];
+      const rows = batch.map((record, index) => {
+        const position = index * 6;
+        values.push(
+          record.sourceId,
+          record.name,
+          JSON.stringify(record.properties),
+          record.longitude,
+          record.latitude,
+          record.properties.fetched_at,
+        );
+        return `(\$${position + 1}, \$${position + 2}, \$${position + 3}::jsonb, \$${position + 4}, \$${position + 5}, \$${position + 6}::timestamptz)`;
+      });
+
+      await client.query(
+        `
+          INSERT INTO marine_features
+            (source_id, name, properties, geom, updated_at, layer, source)
+          SELECT
+            incoming.source_id,
+            incoming.name,
+            incoming.properties,
+            ST_SetSRID(
+              ST_MakePoint(incoming.longitude::double precision, incoming.latitude::double precision),
+              4326
+            ),
+            incoming.updated_at,
+            'locks',
+            'euris'
+          FROM (VALUES ${rows.join(",")}) AS incoming
+            (source_id, name, properties, longitude, latitude, updated_at)
+          ON CONFLICT (source, source_id, layer) WHERE source_id IS NOT NULL
+          DO UPDATE SET
+            name = EXCLUDED.name,
+            properties = EXCLUDED.properties,
+            geom = EXCLUDED.geom,
+            updated_at = EXCLUDED.updated_at
+        `,
+        values,
+      );
+    }
+    await client.query("COMMIT");
+    return records.length;
+  } catch (error) {
+    await client.query("ROLLBACK");
+    throw error;
+  } finally {
+    await client.end();
+  }
+}
+
+export async function runEurisSync({
+  dryRun = false,
+  collectorOptions,
+  writer = writeEurisLocks,
+  databaseUrl,
+} = {}) {
+  const result = await collectEurisLocks(collectorOptions);
+  if (!dryRun) await writer(result.records, { databaseUrl });
+  return { ...result, dryRun };
+}
+
+function boundedInteger(value, fallback, minimum, maximum) {
+  if (value === undefined || value === null || value === "") return fallback;
+  const parsed = Number.parseInt(String(value), 10);
+  if (!Number.isFinite(parsed) || parsed < minimum || parsed > maximum) {
+    throw new Error(`Wert muss zwischen ${minimum} und ${maximum} liegen.`);
+  }
+  return parsed;
+}
+
+async function main() {
+  const countries = parseCountries(process.env.EURIS_COUNTRIES || "DE");
+  const configuredDetailLimit = Number.parseInt(process.env.EURIS_DETAIL_LIMIT || "0", 10);
+  const detailLimit = parseDetailLimit(process.env.EURIS_DETAIL_LIMIT);
+  const dryRun = parseBoolean(process.env.EURIS_DRY_RUN);
+  const timeoutMs = boundedInteger(process.env.EURIS_REQUEST_TIMEOUT_MS, DEFAULT_TIMEOUT_MS, 1_000, 60_000);
+
+  if (Number.isFinite(configuredDetailLimit) && configuredDetailLimit > MAX_DETAIL_LIMIT) {
+    console.warn(`EURIS_DETAIL_LIMIT wurde aus Rücksicht auf das EuRIS-Limit auf ${MAX_DETAIL_LIMIT} begrenzt.`);
+  }
+
+  console.log(
+    `EuRIS-Schleusenabgleich für ${countries.join(", ")} gestartet${dryRun ? " (Trockenlauf)" : ""}.`,
+  );
+  const result = await runEurisSync({
+    dryRun,
+    databaseUrl: process.env.DATABASE_URL,
+    collectorOptions: {
+      countries,
+      detailLimit,
+      token: nonEmptyString(process.env.EURIS_API_TOKEN),
+      timeoutMs,
+    },
+  });
+
+  console.log(JSON.stringify(result.stats, null, 2));
+  console.log(
+    dryRun
+      ? `${result.records.length} Schleusen validiert; die Datenbank wurde nicht verändert.`
+      : `${result.records.length} EuRIS-Schleusen in marine_features aktualisiert.`,
+  );
+}
+
+if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
+  main().catch((error) => {
+    console.error(error);
+    process.exitCode = 1;
+  });
+}