L13 Planung: Geocoding der Objekte über OpenStreetMap Nominatim
Provider-Interface (nominatim|none), strukturierte Suche mit User-Agent und Accept-Language, Drosselung 1/s je Prozess, Job geocode-site (Worker: Concurrency 1 + Limiter), Cache am Objekt, manuelle Koordinaten bleiben, Auslöser Anlage/Adressänderung/Import-Bestätigung nur per Queue, Backfill-Skript, Env-Beispiele. Tests mit Fake-Provider, Nominatim nie aufgerufen. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -18,8 +18,9 @@ async function main() {
|
||||
continue;
|
||||
}
|
||||
const processor = await load();
|
||||
const concurrency = name === JOB_QUEUES.reportPdf ? 2 : 4;
|
||||
const w = new Worker<JobPayload>(name, async (job) => processor(job.data), { connection, concurrency });
|
||||
const geocode = name === JOB_QUEUES.geocodeSite; // L13: OSM Nominatim policy — max. 1 request/s
|
||||
const concurrency = name === JOB_QUEUES.reportPdf ? 2 : geocode ? 1 : 4;
|
||||
const w = new Worker<JobPayload>(name, async (job) => processor(job.data), { connection, concurrency, ...(geocode ? { limiter: { max: 1, duration: 1_000 } } : {}) });
|
||||
w.on("failed", (job, err) => console.error(`[worker] ${name} job ${job?.id} failed:`, err.message));
|
||||
workers.push(w);
|
||||
console.info(`[worker] listening on ${name}`);
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
/**
|
||||
* L13 Planung — geocode sites that have no (or outdated) coordinates, respecting the OSM Nominatim
|
||||
* usage policy: strictly sequential, max. 1 request per second (process-wide limiter in the
|
||||
* provider), identifying User-Agent, results cached on the site (unchanged addresses are skipped).
|
||||
* Manually entered coordinates are never overwritten.
|
||||
*
|
||||
* Usage:
|
||||
* npx tsx scripts/geocode-backfill.ts # all tenants
|
||||
* npx tsx scripts/geocode-backfill.ts --tenant=demo --limit=50
|
||||
* npx tsx scripts/geocode-backfill.ts --dry-run # only list what would be geocoded
|
||||
* npx tsx scripts/geocode-backfill.ts --retry-not-found # also retry addresses marked not_found
|
||||
*
|
||||
* For larger data sets use a self-hosted or contracted geocoding service (GEOCODING_URL).
|
||||
*/
|
||||
import "dotenv/config";
|
||||
import { isValidLatLng } from "../src/lib/geo/distance";
|
||||
import { dbForTenant, prisma } from "../src/server/db";
|
||||
import { geocodingConfig } from "../src/server/services/geo/config";
|
||||
import { geocodeSite, type GeocodeOutcome } from "../src/server/services/geo/geocode-site";
|
||||
import { geocodeQueryFor } from "../src/server/services/geo/normalize";
|
||||
|
||||
const arg = (name: string) => process.argv.find((a) => a.startsWith(`--${name}=`))?.split("=")[1];
|
||||
const flag = (name: string) => process.argv.includes(`--${name}`);
|
||||
|
||||
async function main() {
|
||||
const cfg = geocodingConfig();
|
||||
const dryRun = flag("dry-run");
|
||||
const retryNotFound = flag("retry-not-found");
|
||||
const limit = Number(arg("limit") ?? Infinity);
|
||||
const slug = arg("tenant");
|
||||
if (cfg.provider === "none" && !dryRun) {
|
||||
console.log("GEOCODING_PROVIDER=none — nichts zu tun.");
|
||||
return;
|
||||
}
|
||||
console.log(`Provider ${cfg.provider} (${cfg.url}), User-Agent "${cfg.userAgent}", ≤ 1 Anfrage/s${dryRun ? " — Probelauf" : ""}`);
|
||||
|
||||
const tenants = await prisma.tenant.findMany({ where: slug ? { slug } : {}, select: { id: true, slug: true }, orderBy: { slug: "asc" } });
|
||||
const totals: Record<string, number> = {};
|
||||
let done = 0;
|
||||
for (const tenant of tenants) {
|
||||
const db = dbForTenant(tenant.id);
|
||||
const sites = await db.site.findMany({
|
||||
where: { deletedAt: null },
|
||||
select: { id: true, name: true, street: true, houseNumber: true, postalCode: true, city: true, country: true, latitude: true, longitude: true, geocodedAt: true, geocodeStatus: true, geocodeQuery: true },
|
||||
orderBy: { createdAt: "asc" },
|
||||
});
|
||||
const pending = sites.filter((s) => {
|
||||
const query = geocodeQueryFor(s);
|
||||
if (!query) return false;
|
||||
if (isValidLatLng(s) && !s.geocodedAt) return false; // manual coordinates
|
||||
if (s.geocodeQuery !== query) return true;
|
||||
if (s.geocodeStatus === "ok") return false;
|
||||
if (s.geocodeStatus === "not_found") return retryNotFound;
|
||||
return true;
|
||||
});
|
||||
if (!pending.length) continue;
|
||||
console.log(`\n${tenant.slug}: ${pending.length} Objekt(e)`);
|
||||
for (const site of pending) {
|
||||
if (done >= limit) break;
|
||||
done++;
|
||||
if (dryRun) {
|
||||
console.log(` · ${site.name} (${[site.postalCode, site.city].filter(Boolean).join(" ")})`);
|
||||
continue;
|
||||
}
|
||||
if (retryNotFound && site.geocodeStatus === "not_found") await db.site.update({ where: { id: site.id }, data: { geocodeStatus: null } });
|
||||
const outcome: GeocodeOutcome = await geocodeSite(tenant.id, site.id);
|
||||
totals[outcome] = (totals[outcome] ?? 0) + 1;
|
||||
console.log(` ${outcome.padEnd(9)} ${site.name}`);
|
||||
}
|
||||
}
|
||||
console.log(`\nFertig: ${done} Objekt(e) ${JSON.stringify(totals)}`);
|
||||
}
|
||||
|
||||
main()
|
||||
.catch((err) => {
|
||||
console.error(err);
|
||||
process.exitCode = 1;
|
||||
})
|
||||
.finally(async () => {
|
||||
await prisma.$disconnect();
|
||||
});
|
||||
@@ -0,0 +1,178 @@
|
||||
// L13 Planung — Geocoding: Normalisierung/Änderungserkennung, Nominatim-Anfrage (URL, User-Agent,
|
||||
// Accept-Language, Auswertung, Fehler) mit injiziertem fetch, Drosselung 1/s mit Fake-Uhr, Processor mit
|
||||
// Fake-Provider (Cache bei unveränderter Adresse, not_found, failed → Retry, manuelle Koordinaten,
|
||||
// Provider none, Mandantentrennung), Hook beim Anlegen ohne Inline-Aufruf. Nominatim wird NIE aufgerufen.
|
||||
// Lauf: npx tsx scripts/test-planung-geocode.ts
|
||||
|
||||
import "dotenv/config";
|
||||
import { prisma } from "../src/server/db";
|
||||
import { createTenant, ok, runSuite, section } from "./lib/e2e-fixture";
|
||||
import { geocodingConfig, mapConfig } from "../src/server/services/geo/config";
|
||||
import { requestSiteGeocoding } from "../src/server/services/geo/dispatch";
|
||||
import { geocodeSite } from "../src/server/services/geo/geocode-site";
|
||||
import { buildNominatimUrl, createNominatimProvider } from "../src/server/services/geo/nominatim";
|
||||
import { addressChanged, geocodeQueryFor } from "../src/server/services/geo/normalize";
|
||||
import { GeocodingError, setGeocodingProvider, type GeocodeAddress, type GeocodingProvider } from "../src/server/services/geo/provider";
|
||||
import { createIntervalLimiter } from "../src/server/services/geo/rate-limit";
|
||||
import { process as geocodeProcessor } from "../src/server/jobs/processors/geocode-site";
|
||||
import { PROCESSORS } from "../src/server/jobs/processors";
|
||||
import { createSite } from "../src/server/services/sites/sites";
|
||||
|
||||
const SLUG_A = "zz-planung-geo-a";
|
||||
const SLUG_B = "zz-planung-geo-b";
|
||||
|
||||
// Guard: any request to a Nominatim host fails the suite.
|
||||
let nominatimHits = 0;
|
||||
const realFetch = globalThis.fetch;
|
||||
globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
|
||||
const url = input instanceof Request ? input.url : String(input);
|
||||
if (/nominatim|openstreetmap/i.test(url)) {
|
||||
nominatimHits++;
|
||||
throw new Error("Nominatim must never be called from tests");
|
||||
}
|
||||
return realFetch(input, init);
|
||||
}) as typeof fetch;
|
||||
|
||||
function fakeProvider(answer: (a: GeocodeAddress) => { status: "ok"; latitude: number; longitude: number } | { status: "not_found" } | "throw") {
|
||||
const calls: GeocodeAddress[] = [];
|
||||
const provider: GeocodingProvider = {
|
||||
name: "fake",
|
||||
async geocode(a) {
|
||||
calls.push(a);
|
||||
const r = answer(a);
|
||||
if (r === "throw") throw new GeocodingError("http 503");
|
||||
return r;
|
||||
},
|
||||
};
|
||||
return { provider, calls };
|
||||
}
|
||||
|
||||
runSuite("L13 Planung – Geocoding", [SLUG_A, SLUG_B], async () => {
|
||||
section("Normalisierung & Änderungserkennung");
|
||||
const addr = { street: "Hafenstraße", houseNumber: "12", postalCode: "20457", city: "Hamburg", country: "DE" };
|
||||
ok(geocodeQueryFor(addr) === "hafenstraße 12|20457|hamburg|de", `normierte Adresse (${geocodeQueryFor(addr)})`);
|
||||
ok(geocodeQueryFor({ ...addr, street: " HAFENSTRASSE ".replace("SS", "ß"), city: "hamburg " }) === geocodeQueryFor(addr), "Groß-/Kleinschreibung und Leerzeichen egal");
|
||||
ok(geocodeQueryFor({ street: "Hafenstraße", houseNumber: "12" }) === null, "ohne PLZ und Ort → nicht verortbar");
|
||||
ok(!addressChanged(addr, { ...addr, city: "HAMBURG" }) && addressChanged(addr, { ...addr, houseNumber: "14" }), "nur relevante Adressänderungen zählen");
|
||||
|
||||
section("Nominatim-Anfrage (ohne Netz)");
|
||||
const url = buildNominatimUrl("https://nominatim.example/", addr);
|
||||
ok(
|
||||
url.pathname === "/search" && url.searchParams.get("format") === "jsonv2" && url.searchParams.get("limit") === "1" && url.searchParams.get("street") === "12 Hafenstraße" && url.searchParams.get("postalcode") === "20457" && url.searchParams.get("city") === "Hamburg" && url.searchParams.get("countrycodes") === "de",
|
||||
`strukturierte Suche (${url.search})`,
|
||||
);
|
||||
const cfg = geocodingConfig({ APP_BASE_URL: "https://app.craftvia.example" });
|
||||
ok(/^Craftvia\/\d+\.\d+\.\d+ \(\+https:\/\/app\.craftvia\.example\)$/.test(cfg.userAgent) && cfg.url === "https://nominatim.openstreetmap.org" && cfg.provider === "nominatim", `Standard-User-Agent „${cfg.userAgent}“`);
|
||||
ok(geocodingConfig({ GEOCODING_USER_AGENT: "Betrieb/1 (+mailto:it@example.org)", GEOCODING_PROVIDER: "none", GEOCODING_URL: "https://geo.example/" }).userAgent === "Betrieb/1 (+mailto:it@example.org)", "User-Agent per GEOCODING_USER_AGENT überschreibbar");
|
||||
ok(geocodingConfig({ GEOCODING_PROVIDER: "none" }).provider === "none" && geocodingConfig({ GEOCODING_URL: "https://geo.example/" }).url === "https://geo.example", "Provider none / eigene GEOCODING_URL");
|
||||
ok(mapConfig({}).tileUrl === "https://tile.openstreetmap.org/{z}/{x}/{y}.png" && mapConfig({}).attribution.includes("OpenStreetMap"), "Kartenkacheln: OSM-Standard mit Namensnennung");
|
||||
|
||||
const seen: { url: string; headers: Headers }[] = [];
|
||||
let reply: () => Response = () => Response.json([{ lat: "53.5413", lon: "9.9849" }]);
|
||||
const acquired: number[] = [];
|
||||
const nominatim = createNominatimProvider(
|
||||
{ url: "https://nominatim.example", userAgent: "Craftvia/test (+https://app.example)", timeoutMs: 1000 },
|
||||
{
|
||||
fetch: (async (input: RequestInfo | URL, init?: RequestInit) => {
|
||||
seen.push({ url: String(input), headers: new Headers(init?.headers) });
|
||||
return reply();
|
||||
}) as typeof fetch,
|
||||
limiter: { acquire: async () => (acquired.push(Date.now()), Date.now()) },
|
||||
},
|
||||
);
|
||||
const hit = await nominatim.geocode(addr);
|
||||
ok(hit.status === "ok" && hit.latitude === 53.5413 && hit.longitude === 9.9849, "Treffer ausgewertet");
|
||||
ok(seen[0]?.headers.get("user-agent") === "Craftvia/test (+https://app.example)" && seen[0].headers.get("accept-language") === "de", "User-Agent und Accept-Language de gesetzt");
|
||||
ok(acquired.length === 1, "jede Anfrage läuft über den Drosselungs-Limiter");
|
||||
reply = () => Response.json([]);
|
||||
ok((await nominatim.geocode(addr)).status === "not_found", "leere Antwort → not_found");
|
||||
reply = () => new Response("busy", { status: 503 });
|
||||
let threw = false;
|
||||
try {
|
||||
await nominatim.geocode(addr);
|
||||
} catch (err) {
|
||||
threw = err instanceof GeocodingError;
|
||||
}
|
||||
ok(threw, "HTTP-Fehler → GeocodingError (wiederholbar)");
|
||||
|
||||
section("Drosselung 1 Anfrage/s (Fake-Uhr)");
|
||||
let clock = 10_000;
|
||||
const sleeps: number[] = [];
|
||||
const limiter = createIntervalLimiter(1000, { now: () => clock, sleep: async (ms) => void sleeps.push(ms) });
|
||||
const slots = [await limiter.acquire(), await limiter.acquire(), await limiter.acquire()];
|
||||
ok(slots[0] === 10_000 && slots[1] === 11_000 && slots[2] === 12_000 && sleeps.join(",") === "1000,2000", `Slots im Sekundenabstand (${slots.join(",")}), Wartezeiten ${sleeps.join(",")}`);
|
||||
clock = 20_000;
|
||||
ok((await limiter.acquire()) === 20_000 && sleeps.length === 2, "nach Pause sofort wieder frei");
|
||||
|
||||
section("geocodeSite mit Fake-Provider");
|
||||
const A = await createTenant(SLUG_A);
|
||||
const B = await createTenant(SLUG_B);
|
||||
const fake = fakeProvider((a) => (a.city === "Nirgendwo" ? { status: "not_found" } : { status: "ok", latitude: 53.55, longitude: a.houseNumber === "14" ? 10.01 : 10.0 }));
|
||||
setGeocodingProvider(fake.provider);
|
||||
const site = await prisma.site.create({ data: { tenantId: A.tenantId, customerId: A.customerId, name: "Geo 1", ...addr } });
|
||||
|
||||
ok((await geocodeSite(A.tenantId, site.id)) === "ok" && fake.calls.length === 1, "neue Adresse → ok, 1 Provider-Aufruf");
|
||||
const s1 = await prisma.site.findUniqueOrThrow({ where: { id: site.id } });
|
||||
ok(s1.latitude === 53.55 && s1.longitude === 10.0 && s1.geocodeStatus === "ok" && s1.geocodeQuery === geocodeQueryFor(addr) && !!s1.geocodedAt, "Koordinaten, Status, Adresse und Zeitpunkt am Objekt gespeichert");
|
||||
ok((await prisma.auditLog.count({ where: { tenantId: A.tenantId, entity: "site", entityId: site.id, action: "update" } })) === 1, "Koordinatenänderung auditiert");
|
||||
ok((await geocodeSite(A.tenantId, site.id)) === "cached" && fake.calls.length === 1, "unveränderte Adresse → Cache, kein weiterer Aufruf");
|
||||
await prisma.site.update({ where: { id: site.id }, data: { city: "hamburg " } });
|
||||
ok((await geocodeSite(A.tenantId, site.id)) === "cached" && fake.calls.length === 1, "nur Schreibweise geändert → Cache");
|
||||
await prisma.site.update({ where: { id: site.id }, data: { houseNumber: "14" } });
|
||||
ok((await geocodeSite(A.tenantId, site.id)) === "ok" && fake.calls.length === 2 && (await prisma.site.findUniqueOrThrow({ where: { id: site.id } })).longitude === 10.01, "geänderte Adresse → neu verortet");
|
||||
await prisma.site.update({ where: { id: site.id }, data: { city: "Nirgendwo", postalCode: "99999" } });
|
||||
ok((await geocodeSite(A.tenantId, site.id)) === "not_found" && fake.calls.length === 3, "unbekannte Adresse → not_found");
|
||||
const s2 = await prisma.site.findUniqueOrThrow({ where: { id: site.id } });
|
||||
ok(s2.geocodeStatus === "not_found" && s2.latitude === null && s2.longitude === null, "not_found entfernt veraltete automatische Koordinaten");
|
||||
ok((await geocodeSite(A.tenantId, site.id)) === "cached" && fake.calls.length === 3, "not_found wird für dieselbe Adresse nicht wiederholt");
|
||||
|
||||
const failing = fakeProvider(() => "throw");
|
||||
const siteFail = await prisma.site.create({ data: { tenantId: A.tenantId, customerId: A.customerId, name: "Geo Fehler", ...addr } });
|
||||
ok((await geocodeSite(A.tenantId, siteFail.id, { provider: failing.provider })) === "failed" && (await prisma.site.findUniqueOrThrow({ where: { id: siteFail.id } })).geocodeStatus === "failed", "Providerfehler → Status failed");
|
||||
ok((await geocodeSite(A.tenantId, siteFail.id)) === "ok", "failed wird beim nächsten Lauf erneut versucht");
|
||||
|
||||
const manual = await prisma.site.create({ data: { tenantId: A.tenantId, customerId: A.customerId, name: "Geo manuell", ...addr, latitude: 50.1, longitude: 8.6 } });
|
||||
const callsBefore = fake.calls.length;
|
||||
ok((await geocodeSite(A.tenantId, manual.id)) === "manual" && fake.calls.length === callsBefore, "manuelle Koordinaten → kein Aufruf");
|
||||
const m = await prisma.site.findUniqueOrThrow({ where: { id: manual.id } });
|
||||
ok(m.latitude === 50.1 && m.longitude === 8.6 && m.geocodeStatus === "skipped", "manuelle Koordinaten bleiben unverändert (Status skipped)");
|
||||
|
||||
const none = await prisma.site.create({ data: { tenantId: A.tenantId, customerId: A.customerId, name: "Geo none", ...addr } });
|
||||
ok((await geocodeSite(A.tenantId, none.id, { provider: null })) === "skipped" && (await prisma.site.findUniqueOrThrow({ where: { id: none.id } })).latitude === null, "Provider none → skipped, ohne Ortsangabe, kein Fehler");
|
||||
const noAddr = await prisma.site.create({ data: { tenantId: A.tenantId, customerId: A.customerId, name: "Geo ohne Adresse", street: "Irgendwo" } });
|
||||
ok((await geocodeSite(A.tenantId, noAddr.id)) === "skipped" && fake.calls.length === callsBefore, "unvollständige Adresse → skipped ohne Aufruf");
|
||||
|
||||
const bCalls = fake.calls.length;
|
||||
ok((await geocodeSite(B.tenantId, none.id)) === "missing" && fake.calls.length === bCalls, "Mandant B: Objekt von A nicht gefunden, kein Aufruf");
|
||||
ok((await prisma.site.findUniqueOrThrow({ where: { id: none.id } })).geocodeStatus === "skipped", "Objekt von A unverändert");
|
||||
|
||||
section("Processor & Hook");
|
||||
ok(typeof PROCESSORS["geocode-site"] === "function", "Processor geocode-site registriert");
|
||||
const viaJob = await prisma.site.create({ data: { tenantId: A.tenantId, customerId: A.customerId, name: "Geo Job", ...addr } });
|
||||
await geocodeProcessor({ tenantId: A.tenantId, entityId: viaJob.id });
|
||||
ok((await prisma.site.findUniqueOrThrow({ where: { id: viaJob.id } })).geocodeStatus === "ok", "Processor verortet das Objekt");
|
||||
setGeocodingProvider(failing.provider);
|
||||
const viaJobFail = await prisma.site.create({ data: { tenantId: A.tenantId, customerId: A.customerId, name: "Geo Job Fehler", ...addr } });
|
||||
let jobThrew = false;
|
||||
try {
|
||||
await geocodeProcessor({ tenantId: A.tenantId, entityId: viaJobFail.id });
|
||||
} catch {
|
||||
jobThrew = true;
|
||||
}
|
||||
ok(jobThrew, "Processor wirft bei failed → BullMQ wiederholt");
|
||||
setGeocodingProvider(fake.provider);
|
||||
|
||||
const prev = process.env.GEOCODING_PROVIDER;
|
||||
process.env.GEOCODING_PROVIDER = "none";
|
||||
ok((await requestSiteGeocoding(A.tenantId, viaJob.id)) === false, "GEOCODING_PROVIDER=none → kein Job");
|
||||
process.env.GEOCODING_PROVIDER = "nominatim";
|
||||
const hookCalls = fake.calls.length;
|
||||
const created = await createSite(A.ctx.backoffice, { customerId: A.customerId, name: "Geo Hook", street: "Am Kaiserkai", houseNumber: "1", postalCode: "20457", city: "Hamburg" });
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
ok(!!created.id && fake.calls.length === hookCalls, "Anlage eines Objekts: Verortung nur per Queue, nie inline im Request");
|
||||
if (prev === undefined) delete process.env.GEOCODING_PROVIDER;
|
||||
else process.env.GEOCODING_PROVIDER = prev;
|
||||
setGeocodingProvider(undefined);
|
||||
|
||||
ok(nominatimHits === 0, "Nominatim wurde in keinem Test aufgerufen");
|
||||
});
|
||||
Reference in New Issue
Block a user