From 6751d298d48bd1e25ece191829d5c04e55ad6bcc Mon Sep 17 00:00:00 2001
From: yellowhat <1692490+yellowhat@users.noreply.github.com>
Date: Thu, 17 Sep 2026 21:03:44 +0200
Subject: [PATCH 1/5] test(scrapers): let the registry-alignment guard see
 EV_XX_SOURCE keys

The alignment test checks that every EV_XX scraper key appears in all
three tables: the interval defaults and the factory map in
instrumentation.ts, and SCRAPERS in the CLI. Its regexes matched only
\`EV_\` plus a two-letter country code, so a key with a source suffix such
as EV_ES_REVE stopped matching at \`EV_ES\`, found no \`:\` after it, and was
silently dropped from the comparison. The guard was blind to exactly the
country-specific official registries that are most likely to be added to
one table and forgotten in another.

The regexes now accept an optional _SOURCE suffix. Nothing changes for
the plain EV_XX keys.
---
 src/scrapers/cli.test.ts | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/src/scrapers/cli.test.ts b/src/scrapers/cli.test.ts
index 4b1516d..51c3c78 100644
--- a/src/scrapers/cli.test.ts
+++ b/src/scrapers/cli.test.ts
@@ -33,12 +33,17 @@ function evKeys(source: string, re: RegExp): Set<string> {
 const instrumentation = readSrc("instrumentation.ts");
 const cli = readSrc("scrapers/cli.ts");
 
+// The optional suffix matters: country-specific official registries are keyed
+// EV_XX_SOURCE (EV_ES_REVE, EV_DE_BNETZA). Without it these regexes stop at
+// `EV_ES`, find no `:` after it, and silently drop the key — so the guard was
+// blind to exactly the entries most likely to be added to one table only.
+//
 // EV_XX interval entries: `EV_XX: 24`
-const intervalEvKeys = evKeys(instrumentation, /\b(EV_[A-Z]{2})\s*:\s*\d+/g);
+const intervalEvKeys = evKeys(instrumentation, /\b(EV_[A-Z]{2}(?:_[A-Z0-9]+)?)\s*:\s*\d+/g);
 // EV_XX scraperFactories entries: `EV_XX: () => new OCMScraper(...)`
-const factoryEvKeys = evKeys(instrumentation, /\b(EV_[A-Z]{2})\s*:\s*\(\)\s*=>/g);
+const factoryEvKeys = evKeys(instrumentation, /\b(EV_[A-Z]{2}(?:_[A-Z0-9]+)?)\s*:\s*\(\)\s*=>/g);
 // EV_XX CLI SCRAPERS entries: `EV_XX: [() => new OCMScraper(...)]`
-const cliEvKeys = evKeys(cli, /\b(EV_[A-Z]{2})\s*:\s*\[/g);
+const cliEvKeys = evKeys(cli, /\b(EV_[A-Z]{2}(?:_[A-Z0-9]+)?)\s*:\s*\[/g);
 
 describe("scraper registry alignment", () => {
   it("found EV interval keys to check", () => {

From d3f670cf5fbb49999ed9ec2f6206f10ae4750811 Mon Sep 17 00:00:00 2001
From: yellowhat <1692490+yellowhat@users.noreply.github.com>
Date: Thu, 17 Sep 2026 21:04:45 +0200
Subject: [PATCH 2/5] =?UTF-8?q?feat(ev):=20add=20BNetzA=20Lades=C3=A4ulenr?=
 =?UTF-8?q?egister=20scraper=20for=20Germany?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The Ladesäulenregister is the register every operator of a publicly
accessible charge point in Germany must file into under §5
Ladesäulenverordnung, so it is authoritative where Open Charge Map is
crowdsourced, and several times larger (~74,000 locations). It is
published as one daily bulk TSV, needs no API key and no signup, and is
CC BY 4.0 with the attribution string "Bundesnetzagentur.de".

This commit adds the scraper and its tests only. Nothing imports it yet,
so behaviour is unchanged; the wiring follows in the next commit.

Two things shape the implementation:

Rows are per charging facility, not per location, so one garage is five
identical rows. Those are merged on a 5-decimal (~1 m) coordinate key,
and that same key is the externalId. It has to be the same function:
base.run() upserts in 500-row batches with a single multi-row INSERT ...
ON CONFLICT, which Postgres rejects with a 21000 cardinality violation
when two rows in one statement share a conflict target. On the real file
1,136 coordinates carry two street spellings, so keying the id on
anything coarser than the merge key would mint exactly that many
duplicate ids and silently discard whole batches.

base.run() only orphan-cleans price-less fuel stations, so EV rows are
never garbage-collected. A derived id churns whenever an operator
corrects its coordinates, and each churned id would leak a permanent
pin. run() therefore sweeps rows not touched by the current run, and
retires the Open Charge Map rows for Germany that this source replaces.
Both sweeps are gated on a database-counted floor
(PUMPERLY_BNETZA_MIN_STATIONS, default 10,000) so a degraded fetch can
never delete a working map.

Coordinates are validated with a strict decimal regex rather than
z.coerce.number(), which would turn a blank cell into Null Island, and
rejected outside a padded German bounding box. Columns are resolved by
header name so an inserted column upstream fails the run instead of
swapping latitude and longitude.
---
 src/scrapers/bnetza.test.ts | 530 ++++++++++++++++++++++++++++++++++++
 src/scrapers/bnetza.ts      | 433 +++++++++++++++++++++++++++++
 2 files changed, 963 insertions(+)
 create mode 100644 src/scrapers/bnetza.test.ts
 create mode 100644 src/scrapers/bnetza.ts

diff --git a/src/scrapers/bnetza.test.ts b/src/scrapers/bnetza.test.ts
new file mode 100644
index 0000000..b648714
--- /dev/null
+++ b/src/scrapers/bnetza.test.ts
@@ -0,0 +1,530 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+
+vi.mock("@prisma/adapter-pg", () => ({ PrismaPg: vi.fn() }));
+vi.mock("../generated/prisma/client", () => ({ PrismaClient: vi.fn() }));
+
+// The register's 26 columns, in file order.
+const HEADER = [
+  "Betreiber",
+  "Adresszusatz",
+  "Straße",
+  "Hausnummer",
+  "Postleitzahl",
+  "Ort",
+  "Breitengrad",
+  "Längengrad",
+  "Inbetriebnahmedatum",
+  "Nennleistung Ladeeinrichtung [kW]",
+  "Art der Ladeeinrichtung",
+  "Anzahl Ladepunkte",
+  "Steckertypen1",
+  "Nennleistung Stecker1",
+  "Steckertypen2",
+  "Nennleistung Stecker2",
+  "Steckertypen3",
+  "Nennleistung Stecker3",
+  "Steckertypen4",
+  "Nennleistung Stecker4",
+  "Zugangsbeschränkung",
+  "Öffnungszeiten",
+  "Debitkarte",
+  "Kreditkarte",
+  "Bargeld",
+  "Status",
+] as const;
+
+type Field = (typeof HEADER)[number];
+
+// One row shaped exactly like a real register entry (the Berliner Stadtwerke
+// charger on Leipziger Platz), overridable per test.
+function row(overrides: Partial<Record<Field, string>> = {}): string {
+  const base: Record<string, string> = {
+    Betreiber: "Berliner Stadtwerke KommunalPartner GmbH",
+    Adresszusatz: "",
+    "Straße": "Leipziger Platz",
+    Hausnummer: "19",
+    Postleitzahl: "01011",
+    Ort: "Berlin",
+    Breitengrad: "52.510055",
+    "Längengrad": "13.377592",
+    Inbetriebnahmedatum: "30.06.2023",
+    "Nennleistung Ladeeinrichtung [kW]": "30.0",
+    "Art der Ladeeinrichtung": "0",
+    "Anzahl Ladepunkte": "2",
+    Steckertypen1: "AC Typ 2 Steckdose",
+    "Nennleistung Stecker1": "22",
+    Steckertypen2: "AC Typ 2 Steckdose",
+    "Nennleistung Stecker2": "22",
+    Steckertypen3: "",
+    "Nennleistung Stecker3": "",
+    Steckertypen4: "",
+    "Nennleistung Stecker4": "",
+    "Zugangsbeschränkung": "",
+    "Öffnungszeiten": "247",
+    Debitkarte: "",
+    Kreditkarte: "",
+    Bargeld: "",
+    Status: "1",
+    ...overrides,
+  };
+  return HEADER.map((h) => base[h]).join("\t");
+}
+
+function tsv(...rows: string[]): string {
+  return [HEADER.join("\t"), ...rows].join("\n");
+}
+
+describe("parseBnetzaTsv", () => {
+  it("maps an operational row into an EV charger station", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const { stations, stats } = parseBnetzaTsv(tsv(row()));
+
+    expect(stations).toHaveLength(1);
+    expect(stations[0]).toEqual({
+      externalId: "bnetza-52.51006_13.37759",
+      name: "Berliner Stadtwerke KommunalPartner GmbH — Leipziger Platz 19",
+      brand: "Berliner Stadtwerke KommunalPartner GmbH",
+      address: "Leipziger Platz 19, 01011",
+      city: "Berlin",
+      province: null,
+      latitude: 52.510055,
+      longitude: 13.377592,
+      stationType: "ev_charger",
+    });
+    expect(stats).toEqual({
+      totalRows: 1,
+      malformed: 0,
+      notOperational: 0,
+      outOfBbox: 0,
+      mergedDuplicates: 0,
+      operatorConflicts: 0,
+    });
+  });
+
+  it("carries both Adresszusatz and the street in the search name", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const { stations } = parseBnetzaTsv(
+      tsv(row({ Betreiber: "Q-Park Recharge Germany GmbH", Adresszusatz: "Tiefgarage" })),
+    );
+    // "Tiefgarage" alone appears on hundreds of rows and disambiguates nothing;
+    // the street is what tells one Q-Park garage from the next.
+    expect(stations[0].name).toBe("Q-Park Recharge Germany GmbH — Tiefgarage, Leipziger Platz 19");
+    expect(stations[0].address).toBe("Leipziger Platz 19, 01011");
+  });
+
+  it("merges the per-Ladeeinrichtung rows of one site into a single station", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    // The real Q-Park garage at Landhausstraße 2, Dresden: five identical rows.
+    const site = {
+      Betreiber: "Q-Park Recharge Germany GmbH",
+      "Straße": "Landhausstraße",
+      Hausnummer: "2",
+      Breitengrad: "51.050944",
+      "Längengrad": "13.74154",
+    };
+    const { stations, stats } = parseBnetzaTsv(
+      tsv(row(site), row(site), row(site), row(site), row(site)),
+    );
+
+    expect(stations).toHaveLength(1);
+    expect(stats.totalRows).toBe(5);
+    expect(stats.mergedDuplicates).toBe(4);
+    expect(stations[0].externalId).toBe("bnetza-51.05094_13.74154");
+  });
+
+  it("merges within ~1m but keeps distinct chargers apart", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const at = (lat: string) => row({ Breitengrad: lat, "Längengrad": "13.74154" });
+
+    // Sub-metre apart: the same physical site published at two precisions.
+    expect(parseBnetzaTsv(tsv(at("51.0500014"), at("51.0500042"))).stations).toHaveLength(1);
+    // ~11m apart: genuinely different chargers, which 4 decimals would fold.
+    expect(parseBnetzaTsv(tsv(at("51.0501"), at("51.0502"))).stations).toHaveLength(2);
+  });
+
+  it("gives one station one id when coordinates repeat under different street spellings", async () => {
+    // The regression guard for the batch-upsert collision: 1,136 coordinate
+    // pairs in the real file carry two street spellings, so an externalId
+    // derived on anything coarser than the merge key mints duplicate ids and
+    // Postgres rejects the whole 500-row batch with 21000.
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const { stations } = parseBnetzaTsv(
+      tsv(
+        row({ "Straße": "Landhausstr.", Breitengrad: "51.050944", "Längengrad": "13.74154" }),
+        row({ "Straße": "Landhausstraße", Breitengrad: "51.050944", "Längengrad": "13.74154" }),
+      ),
+    );
+    expect(stations).toHaveLength(1);
+    expect(stations[0].address).toBe("Landhausstr. 19, 01011"); // first row wins
+  });
+
+  it("emits no duplicate externalIds", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const { stations } = parseBnetzaTsv(
+      tsv(
+        row(),
+        row(),
+        row({ "Straße": "Anderer Weg" }),
+        row({ Breitengrad: "51.438147", "Längengrad": "14.244672" }),
+        row({ Breitengrad: "51.438147", "Längengrad": "14.244672", Betreiber: "Andere GmbH" }),
+      ),
+    );
+    const ids = stations.map((s) => s.externalId);
+    expect(new Set(ids).size).toBe(ids.length);
+  });
+
+  it("counts an operator disagreement when merging, as a signal the key over-merges", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const { stations, stats } = parseBnetzaTsv(
+      tsv(row({ Betreiber: "EnBW mobility+ AG und Co.KG" }), row({ Betreiber: "E.ON Drive GmbH" })),
+    );
+    expect(stations).toHaveLength(1);
+    expect(stats.mergedDuplicates).toBe(1);
+    expect(stats.operatorConflicts).toBe(1);
+  });
+
+  it("drops non-operational rows without calling them malformed", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const { stations, stats } = parseBnetzaTsv(tsv(row({ Status: "0" }), row()));
+
+    expect(stations).toHaveLength(1);
+    expect(stats.notOperational).toBe(1);
+    expect(stats.malformed).toBe(0);
+  });
+
+  it("drops coordinates outside Germany", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    // The real bad row: a Lemgo charger published at longitude 4.
+    const { stations, stats } = parseBnetzaTsv(
+      tsv(
+        row({
+          Betreiber: "Wirelane GmbH",
+          "Straße": "Lagesche Straße",
+          Hausnummer: "32",
+          Ort: "Lemgo",
+          Breitengrad: "52.023038",
+          "Längengrad": "4",
+        }),
+        row(),
+      ),
+    );
+
+    expect(stations).toHaveLength(1);
+    expect(stats.outOfBbox).toBe(1);
+    expect(stats.malformed).toBe(0);
+  });
+
+  it("drops malformed rows and keeps their neighbours", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const shortRow = HEADER.slice(1)
+      .map(() => "x")
+      .join("\t"); // 25 fields
+    const { stations, stats } = parseBnetzaTsv(
+      tsv(row(), shortRow, row({ Breitengrad: "n/a", "Längengrad": "13.0" }), row({ Ort: "Köln" })),
+    );
+
+    expect(stats.malformed).toBe(2);
+    expect(stations).toHaveLength(1); // rows 1 and 4 share coordinates → merged
+    expect(stations[0].city).toBe("Berlin");
+  });
+
+  it("rejects a blank coordinate rather than placing it at Null Island", async () => {
+    // z.coerce.number() would turn "" into 0 and put this charger in the
+    // Atlantic; the schema's regex rejects it outright.
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const { stations, stats } = parseBnetzaTsv(
+      tsv(row({ Breitengrad: "", "Längengrad": "" }), row({ Breitengrad: "52,5", "Längengrad": "13,3" })),
+    );
+
+    expect(stations).toEqual([]);
+    expect(stats.malformed).toBe(2);
+  });
+
+  it("unquotes RFC4180 fields", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const { stations } = parseBnetzaTsv(
+      tsv(
+        row({
+          Betreiber: '"Hotel ""Zur Mühle"" GmbH"',
+          Adresszusatz: '"Parkplatz ""Am Neumarkt am Wehr"""',
+        }),
+      ),
+    );
+
+    expect(stations[0].brand).toBe('Hotel "Zur Mühle" GmbH');
+    expect(stations[0].name).toBe(
+      'Hotel "Zur Mühle" GmbH — Parkplatz "Am Neumarkt am Wehr", Leipziger Platz 19',
+    );
+  });
+
+  it("skips blank and trailing lines without counting them", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const { stations, stats } = parseBnetzaTsv(`${tsv(row())}\n\n   \n`);
+
+    expect(stations).toHaveLength(1);
+    expect(stats.totalRows).toBe(1);
+    expect(stats.malformed).toBe(0);
+  });
+
+  it("never emits the header as a station", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const { stations, stats } = parseBnetzaTsv(tsv());
+
+    expect(stations).toEqual([]);
+    expect(stats.totalRows).toBe(0);
+  });
+
+  it("throws, naming the column, when the header changes shape", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const renamed = tsv(row()).replace("Breitengrad", "Latitude");
+
+    // Throwing is the point: base.run() catches it and does nothing
+    // destructive, so a format change costs a cycle rather than the table.
+    expect(() => parseBnetzaTsv(renamed)).toThrow(/Breitengrad/);
+  });
+
+  it("parses CRLF input identically to LF", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    const lf = tsv(row());
+    const crlf = lf.replace(/\n/g, "\r\n");
+
+    // Without \r handling the trailing Status column becomes "1\r" and every
+    // row would silently look non-operational.
+    expect(parseBnetzaTsv(crlf).stations).toEqual(parseBnetzaTsv(lf).stations);
+    expect(parseBnetzaTsv(crlf).stations).toHaveLength(1);
+  });
+
+  it("tolerates a UTF-8 BOM on the header", async () => {
+    const { parseBnetzaTsv } = await import("./bnetza");
+    expect(parseBnetzaTsv(`${tsv(row())}`).stations).toHaveLength(1);
+  });
+});
+
+describe("stationKey", () => {
+  it("is the externalId, so both round identically", async () => {
+    const { stationKey, parseBnetzaTsv } = await import("./bnetza");
+    const { stations } = parseBnetzaTsv(tsv(row()));
+    expect(stations[0].externalId).toBe(stationKey(52.510055, 13.377592));
+  });
+
+  it("is URL-safe, since externalIds go into share links", async () => {
+    const { stationKey } = await import("./bnetza");
+    const key = stationKey(52.510055, 13.377592);
+    expect(encodeURIComponent(key)).toBe(key);
+  });
+});
+
+describe("unquote", () => {
+  it("leaves ordinary values alone and collapses doubled quotes", async () => {
+    const { unquote } = await import("./bnetza");
+    expect(unquote("EnBW mobility+ AG und Co.KG")).toBe("EnBW mobility+ AG und Co.KG");
+    expect(unquote("  padded  ")).toBe("padded");
+    expect(unquote('""')).toBe("");
+    expect(unquote('"')).toBe('"'); // too short to be a quoted field
+    expect(unquote('"Parkplatz ""Am Neumarkt am Wehr"""')).toBe('Parkplatz "Am Neumarkt am Wehr"');
+  });
+});
+
+describe("BNetzAScraper", () => {
+  beforeEach(() => {
+    vi.stubGlobal("fetch", vi.fn());
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+    vi.resetModules();
+    vi.unstubAllEnvs();
+  });
+
+  function okResponse(body: string) {
+    return {
+      ok: true,
+      status: 200,
+      text: async () => body,
+    } as unknown as Response;
+  }
+
+  it("has correct source and country", async () => {
+    const { BNetzAScraper } = await import("./bnetza");
+    const scraper = new BNetzAScraper();
+    expect(scraper.country).toBe("DE");
+    expect(scraper.source).toBe("bnetza");
+  });
+
+  it("returns stations and never invents a price", async () => {
+    const { BNetzAScraper } = await import("./bnetza");
+    vi.mocked(fetch).mockResolvedValue(okResponse(tsv(row(), row({ Ort: "Köln" }))));
+
+    const { stations, prices } = await new BNetzAScraper().fetch();
+
+    expect(prices).toEqual([]);
+    expect(stations).toHaveLength(1);
+    expect(stations[0].stationType).toBe("ev_charger");
+  });
+
+  it("identifies itself and bounds the download", async () => {
+    const { BNetzAScraper } = await import("./bnetza");
+    vi.mocked(fetch).mockResolvedValue(okResponse(tsv(row())));
+
+    await new BNetzAScraper().fetch();
+
+    const [url, init] = vi.mocked(fetch).mock.calls[0];
+    expect(String(url)).toBe("https://lade.info/data/stationen_XXXX.txt");
+    const opts = init as RequestInit & { headers: Record<string, string> };
+    expect(opts.headers["User-Agent"]).toMatch(/^Pumperly\//);
+    expect(opts.signal).toBeInstanceOf(AbortSignal);
+  });
+
+  it("throws on a non-OK response rather than wiping data", async () => {
+    const { BNetzAScraper } = await import("./bnetza");
+    vi.mocked(fetch).mockResolvedValue({
+      ok: false,
+      status: 503,
+      text: async () => "upstream down",
+    } as unknown as Response);
+
+    await expect(new BNetzAScraper().fetch()).rejects.toThrow(/503/);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// The cleanup sweeps, which are the only destructive thing this scraper does.
+// base.run() is stubbed out so these exercise the gating and the SQL alone.
+// ---------------------------------------------------------------------------
+describe("BNetzAScraper cleanup", () => {
+  afterEach(() => {
+    vi.restoreAllMocks();
+    vi.resetModules();
+    vi.unstubAllEnvs();
+  });
+
+  /** A PrismaClient stand-in that records every statement it is handed. */
+  function stubPrisma(results: Array<Array<{ count: bigint }>>) {
+    const queries: Array<{ sql: string; params: unknown[] }> = [];
+    const client = {
+      $queryRawUnsafe: vi.fn(async (sql: string, ...params: unknown[]) => {
+        queries.push({ sql, params });
+        return results.shift() ?? [];
+      }),
+      $disconnect: vi.fn(async () => {}),
+    };
+    return { queries, client };
+  }
+
+  async function setup(opts: {
+    results?: Array<Array<{ count: bigint }>>;
+    errors?: string[];
+    durationMs?: number;
+  }) {
+    const { PrismaClient } = await import("../generated/prisma/client");
+    const { BaseScraper } = await import("./base");
+    const stub = stubPrisma(opts.results ?? []);
+    // A plain function, not an arrow: this stands in for a constructor.
+    vi.mocked(PrismaClient).mockImplementation(function () {
+      return stub.client;
+    } as never);
+    vi.spyOn(BaseScraper.prototype, "run").mockResolvedValue({
+      country: "DE",
+      source: "bnetza",
+      stationsUpserted: 74223,
+      pricesUpserted: 0,
+      durationMs: opts.durationMs ?? 30_000,
+      errors: opts.errors ?? [],
+    });
+    return stub;
+  }
+
+  it("prunes stale rows and retires OpenChargeMap once the floor is cleared", async () => {
+    const stub = await setup({
+      results: [[{ count: BigInt(74223) }], [{ count: BigInt(12) }], [{ count: BigInt(19067) }]],
+    });
+    const { BNetzAScraper } = await import("./bnetza");
+
+    const result = await new BNetzAScraper().run();
+
+    expect(result.errors).toEqual([]);
+    expect(stub.queries).toHaveLength(3);
+    expect(stub.queries[0].sql).toMatch(/count\(\*\)/);
+
+    // Stale sweep: scoped to this source, and cut off by a duration measured
+    // back from the DB clock so app/DB clock skew cannot delete fresh rows.
+    expect(stub.queries[1].sql).toMatch(/DELETE FROM stations/);
+    expect(stub.queries[1].sql).toMatch(/external_id LIKE 'bnetza-%'/);
+    expect(stub.queries[1].sql).toMatch(/updated_at < NOW\(\) - make_interval/);
+    expect(stub.queries[1].params[0]).toBe(30 + 300); // run duration + margin
+
+    // Retirement: OpenChargeMap's German chargers, which nothing else removes.
+    expect(stub.queries[2].sql).toMatch(/external_id LIKE 'ocm-%'/);
+    expect(stub.queries[2].sql).toMatch(/station_type = 'ev_charger'/);
+  });
+
+  it("deletes nothing when too few stations were stored", async () => {
+    // A degraded fetch that still clears base.run()'s empty-fetch guard must
+    // not be able to wipe yesterday's map.
+    const stub = await setup({ results: [[{ count: BigInt(42) }]] });
+    const { BNetzAScraper } = await import("./bnetza");
+
+    await new BNetzAScraper().run();
+
+    expect(stub.queries).toHaveLength(1);
+    expect(stub.queries[0].sql).not.toMatch(/DELETE/);
+  });
+
+  it("honours PUMPERLY_BNETZA_MIN_STATIONS", async () => {
+    vi.stubEnv("PUMPERLY_BNETZA_MIN_STATIONS", "40");
+    const stub = await setup({
+      results: [[{ count: BigInt(42) }], [{ count: BigInt(0) }], [{ count: BigInt(0) }]],
+    });
+    const { BNetzAScraper } = await import("./bnetza");
+
+    await new BNetzAScraper().run();
+
+    expect(stub.queries).toHaveLength(3); // 42 now clears the lowered floor
+  });
+
+  it("ignores a nonsensical floor rather than disabling the guard", async () => {
+    vi.stubEnv("PUMPERLY_BNETZA_MIN_STATIONS", "not-a-number");
+    const stub = await setup({ results: [[{ count: BigInt(42) }]] });
+    const { BNetzAScraper } = await import("./bnetza");
+
+    await new BNetzAScraper().run();
+
+    expect(stub.queries).toHaveLength(1); // fell back to the 10,000 default
+  });
+
+  it("does not clean up after a run that reported errors", async () => {
+    const stub = await setup({ errors: ["Station batch 0-500: boom"] });
+    const { BNetzAScraper } = await import("./bnetza");
+
+    const result = await new BNetzAScraper().run();
+
+    expect(stub.queries).toHaveLength(0);
+    expect(result.errors).toEqual(["Station batch 0-500: boom"]);
+  });
+
+  it("reports a cleanup failure instead of throwing out of run()", async () => {
+    const { PrismaClient } = await import("../generated/prisma/client");
+    const { BaseScraper } = await import("./base");
+    vi.mocked(PrismaClient).mockImplementation(function () {
+      return {
+        $queryRawUnsafe: vi.fn(async () => {
+          throw new Error("connection reset");
+        }),
+        $disconnect: vi.fn(async () => {}),
+      };
+    } as never);
+    vi.spyOn(BaseScraper.prototype, "run").mockResolvedValue({
+      country: "DE",
+      source: "bnetza",
+      stationsUpserted: 74223,
+      pricesUpserted: 0,
+      durationMs: 1000,
+      errors: [],
+    });
+    const { BNetzAScraper } = await import("./bnetza");
+
+    const result = await new BNetzAScraper().run();
+
+    expect(result.errors).toEqual(["Cleanup: connection reset"]);
+  });
+});
diff --git a/src/scrapers/bnetza.ts b/src/scrapers/bnetza.ts
new file mode 100644
index 0000000..621b33c
--- /dev/null
+++ b/src/scrapers/bnetza.ts
@@ -0,0 +1,433 @@
+import { z } from "zod";
+import { PrismaPg } from "@prisma/adapter-pg";
+import { PrismaClient } from "../generated/prisma/client";
+import { BaseScraper, type RawFuelPrice, type RawStation, type ScraperResult } from "./base";
+
+// ---------------------------------------------------------------------------
+// BNetzA Ladesäulenregister — official German EV charging point registry
+// ---------------------------------------------------------------------------
+// File:    https://lade.info/data/stationen_XXXX.txt  (~19 MB TSV, daily)
+// Source:  Bundesnetzagentur. Every operator of a publicly accessible charge
+//          point must file it under §5 Ladesäulenverordnung, so this is the
+//          authoritative German register where OpenChargeMap is crowdsourced.
+// Licence: CC BY 4.0. Required attribution string: "Bundesnetzagentur.de"
+//          (see components/nav/legal-modal.tsx). No API key, no signup.
+//
+// Caveat worth knowing: the register only publishes operators who completed the
+// notification procedure AND consented to publication, so it is not exhaustive.
+// It is still several times larger than OpenChargeMap's German coverage.
+//
+// TWO THINGS SHAPE THIS FILE.
+//
+// 1. Rows are per-Ladeeinrichtung, not per-location. The Q-Park garage at
+//    Landhausstraße 2 in Dresden is five identical rows. Left alone those
+//    become five map pins — and worse, `base.ts` upserts in 500-row batches
+//    with a single multi-row `INSERT ... ON CONFLICT DO UPDATE`, which
+//    Postgres rejects with 21000 CARDINALITY_VIOLATION when two rows in the
+//    same statement hit the same conflict target. `base.run()` catches that
+//    per batch, so one duplicate silently discards 499 good stations and
+//    disables orphan cleanup for the run. It only fires when the duplicates
+//    land in the same slice, so it fails intermittently rather than loudly.
+//    Hence: merge before returning, and make the merge key and the externalId
+//    THE SAME FUNCTION. Deriving the ID from anything coarser than the merge
+//    key re-introduces the collision. (Measured on the real file: 74,223
+//    unique 5-decimal coordinate pairs but 75,359 coordinate+street pairs —
+//    1,136 coordinates carry two different street spellings, so keying the ID
+//    on coordinates while merging on coordinates+street would mint exactly
+//    1,136 duplicate IDs.)
+//
+//    Note this departs from the repo's other dedupes (ocm.ts, germany.ts,
+//    argentina.ts) — those collapse overlapping fetch windows and take
+//    first-or-last. This one merges rows that are genuinely distinct upstream.
+//
+// 2. `base.run()`'s orphan cleanup only ever deletes `station_type = 'fuel'`,
+//    so EV rows are never garbage-collected by the framework. A derived ID
+//    churns whenever an operator corrects its coordinates, and every churned
+//    ID would leak a permanent pin. Hence the staleness sweep in run().
+// ---------------------------------------------------------------------------
+
+const BASE_URL = "https://lade.info/data/stationen_XXXX.txt";
+
+// Overall deadline, not an idle timeout: a download still making progress is
+// killed anyway when it expires. ~19 MB uncompressed, ~3-4 MB on the wire once
+// undici negotiates gzip, so this is many times more headroom than needed.
+const DOWNLOAD_TIMEOUT_MS = 300_000;
+
+// Germany's bounding box, generously padded. Catches coordinate typos that are
+// syntactically fine but geographically impossible — the register currently
+// holds one (a Lemgo charger published at longitude 4, true value ~8.9).
+const DE_BBOX = { latMin: 47, latMax: 56, lonMin: 5.5, lonMax: 15.5 };
+
+// Merge granularity, in decimal places. 5 dp is ~1.1 m, which collapses the
+// per-Ladeeinrichtung rows of one site without merging neighbours. Measured
+// alternatives on the real file: 4 dp yields 70,236 stations and demonstrably
+// folds together distinct chargers; raw unrounded strings yield 76,441 and
+// leave sub-metre duplicates behind. This value is baked into every externalId
+// ever written, so changing it re-mints ~74k rows — treat it as a migration.
+const COORD_PRECISION = 5;
+
+// Floor gating the two destructive sweeps in run(). A degraded fetch that still
+// clears the empty-fetch guard must not be able to retire OpenChargeMap or bulk
+// delete yesterday's stations. The register holds ~74k; 10k is a wide margin.
+const rawMinStations = Number(process.env.PUMPERLY_BNETZA_MIN_STATIONS ?? "10000");
+const MIN_STATIONS =
+  Number.isFinite(rawMinStations) && rawMinStations > 0 ? Math.floor(rawMinStations) : 10_000;
+
+// Slack added to the staleness cutoff so the sweep can never delete a row
+// written in the opening moments of the run it belongs to.
+const SWEEP_MARGIN_SECONDS = 300;
+
+// Share of merged rows that may disagree on operator before it is worth
+// flagging. Measured at ~1.2% on the real file, which is colocated chargers run
+// by different companies, not a merge fault.
+const OPERATOR_CONFLICT_WARN_RATIO = 0.05;
+
+// Columns consumed, by header name. Resolved by name rather than fixed index so
+// that an inserted column upstream cannot silently shift latitude into
+// longitude — the failure mode of positional parsing is bad data, not an error.
+const COLUMNS = {
+  operator: "Betreiber",
+  addressExtra: "Adresszusatz",
+  street: "Straße",
+  houseNumber: "Hausnummer",
+  postcode: "Postleitzahl",
+  city: "Ort",
+  latitude: "Breitengrad",
+  longitude: "Längengrad",
+  status: "Status",
+} as const;
+
+type ColumnIndices = Record<keyof typeof COLUMNS, number>;
+
+// Every field is a `string` by construction after splitting on tabs, so a
+// shape-only schema here would be incapable of failing. Zod earns its place by
+// owning the string→number conversion instead.
+//
+// Deliberately NOT z.coerce.number(): under zod 4 it maps "" and null to 0,
+// which would manufacture a coordinate at Null Island out of a blank cell. The
+// regex rejects "", " ", "8,9" and "abc" alike.
+const decimal = z
+  .string()
+  .regex(/^-?\d+(\.\d+)?$/, "expected a decimal number")
+  .transform(Number);
+
+const CoordinatesSchema = z.object({
+  latitude: decimal,
+  longitude: decimal,
+});
+
+export interface BnetzaParseStats {
+  /** Non-blank data rows seen (excludes the header). */
+  totalRows: number;
+  /** Wrong field count, or coordinates that are not decimal numbers. */
+  malformed: number;
+  /** Status column is not "1". A business filter, not a parse failure. */
+  notOperational: number;
+  /** Syntactically valid coordinates that fall outside Germany. */
+  outOfBbox: number;
+  /** Rows folded into a station already seen. */
+  mergedDuplicates: number;
+  /** Merged rows whose operator disagreed — a signal the key over-merges. */
+  operatorConflicts: number;
+}
+
+/**
+ * A handful of fields use RFC4180 quoting (`"Hotel ""Zur Mühle"" GmbH"`).
+ * Those fields never contain tabs or newlines, so splitting the file remains
+ * safe and this is only needed to clean up the values for display.
+ */
+export function unquote(field: string): string {
+  const s = field.trim();
+  if (s.length >= 2 && s.startsWith('"') && s.endsWith('"')) {
+    return s.slice(1, -1).replace(/""/g, '"').trim();
+  }
+  return s;
+}
+
+/**
+ * The merge key, which is also the externalId. See the header comment: these
+ * must be one and the same function or the batch upsert breaks.
+ *
+ * Exported so a test can pin the merge granularity.
+ */
+export function stationKey(latitude: number, longitude: number): string {
+  return `bnetza-${latitude.toFixed(COORD_PRECISION)}_${longitude.toFixed(COORD_PRECISION)}`;
+}
+
+/**
+ * Map header names to column positions.
+ *
+ * Throws when a column is missing or renamed. That is deliberate: `base.run()`
+ * catches it and returns a no-op run, so an upstream format change costs a
+ * scrape cycle instead of corrupting the table.
+ */
+function resolveColumns(headerLine: string): ColumnIndices {
+  const header = headerLine
+    .replace(/^/, "")
+    .split("\t")
+    .map((h) => unquote(h).toLowerCase());
+
+  const indices = {} as ColumnIndices;
+  for (const [field, name] of Object.entries(COLUMNS) as Array<
+    [keyof typeof COLUMNS, string]
+  >) {
+    const idx = header.indexOf(name.toLowerCase());
+    if (idx === -1) {
+      throw new Error(`BNetzA: column "${name}" missing from header`);
+    }
+    indices[field] = idx;
+  }
+  return indices;
+}
+
+/**
+ * Parse the register TSV into deduplicated stations.
+ *
+ * Pure and exported: every filter, the merge and the derived externalId live
+ * here, so this is the only part that needs testing and it needs no network.
+ */
+export function parseBnetzaTsv(text: string): {
+  stations: RawStation[];
+  stats: BnetzaParseStats;
+} {
+  // `\r?\n` rather than `\n`: the file is LF today, but a CRLF republish would
+  // otherwise append a stray \r to the last column and break the status filter.
+  const lines = text.split(/\r?\n/);
+  const cols = resolveColumns(lines[0] ?? "");
+  const fieldCount = (lines[0] ?? "").split("\t").length;
+
+  const stats: BnetzaParseStats = {
+    totalRows: 0,
+    malformed: 0,
+    notOperational: 0,
+    outOfBbox: 0,
+    mergedDuplicates: 0,
+    operatorConflicts: 0,
+  };
+
+  const byKey = new Map<string, RawStation>();
+
+  for (let i = 1; i < lines.length; i++) {
+    const line = lines[i];
+    if (line.trim() === "") continue; // blank and trailing lines are not errors
+    stats.totalRows++;
+
+    // Split inside the loop and discard. `lines.map(l => l.split("\t"))` would
+    // hold ~117k arrays alive at once (~125 MB) — that, not the 19 MB download,
+    // is what would OOM the 512 MB app container.
+    const fields = line.split("\t");
+    if (fields.length !== fieldCount) {
+      stats.malformed++;
+      continue;
+    }
+
+    if (unquote(fields[cols.status]) !== "1") {
+      stats.notOperational++;
+      continue;
+    }
+
+    const parsed = CoordinatesSchema.safeParse({
+      latitude: unquote(fields[cols.latitude]),
+      longitude: unquote(fields[cols.longitude]),
+    });
+    if (!parsed.success) {
+      stats.malformed++;
+      continue;
+    }
+    const { latitude, longitude } = parsed.data;
+
+    // Range checks stay out of the schema, matching ocm.ts and reve.ts: an
+    // implausible coordinate is a value problem, not a shape problem, and
+    // conflating the two makes the "malformed" counter useless as a signal.
+    if (
+      latitude < DE_BBOX.latMin ||
+      latitude > DE_BBOX.latMax ||
+      longitude < DE_BBOX.lonMin ||
+      longitude > DE_BBOX.lonMax
+    ) {
+      stats.outOfBbox++;
+      continue;
+    }
+
+    const key = stationKey(latitude, longitude);
+    const operator = unquote(fields[cols.operator]);
+
+    const existing = byKey.get(key);
+    if (existing) {
+      stats.mergedDuplicates++;
+      if (operator && existing.brand && operator !== existing.brand) {
+        stats.operatorConflicts++;
+      }
+      continue;
+    }
+
+    const street = unquote(fields[cols.street]);
+    const houseNumber = unquote(fields[cols.houseNumber]);
+    const addressExtra = unquote(fields[cols.addressExtra]);
+    const postcode = unquote(fields[cols.postcode]);
+    const streetLine = [street, houseNumber].filter(Boolean).join(" ");
+
+    // `name` is the search fallback and the small grey line in the results
+    // list, not the popup heading (that shows brand + address). Carry both
+    // Adresszusatz and the street: Adresszusatz is present on only ~18% of rows
+    // and is as often a bare "Parkplatz" or an asset code ("T175-IT1-1322-212")
+    // as a real site label, so on its own it disambiguates nothing — while the
+    // street alone loses the one useful case, two chargers at one address.
+    const label = [addressExtra, streetLine].filter(Boolean).join(", ");
+    const name = [operator, label].filter(Boolean).join(" — ") || key;
+
+    byKey.set(key, {
+      externalId: key,
+      name,
+      brand: operator || null,
+      address: [streetLine, postcode].filter(Boolean).join(", ") || name,
+      city: unquote(fields[cols.city]),
+      // No Bundesland column, and PLZ→Bundesland is not a function: 04xxx,
+      // 34xxx, 37xxx and 49xxx each straddle Land borders. Nothing in the app
+      // renders province, and germany.ts already writes null for DE.
+      province: null,
+      latitude,
+      longitude,
+      stationType: "ev_charger",
+    });
+  }
+
+  return { stations: [...byKey.values()], stats };
+}
+
+export class BNetzAScraper extends BaseScraper {
+  readonly country = "DE";
+  readonly source = "bnetza";
+
+  async fetch(): Promise<{ stations: RawStation[]; prices: RawFuelPrice[] }> {
+    const res = await fetch(BASE_URL, {
+      headers: {
+        Accept: "text/plain",
+        "User-Agent": "Pumperly/1.0 (+https://pumperly.com)",
+      },
+      signal: AbortSignal.timeout(DOWNLOAD_TIMEOUT_MS),
+    });
+
+    if (!res.ok) {
+      throw new Error(`BNetzA HTTP ${res.status}: ${await res.text().catch(() => "")}`);
+    }
+
+    // ~19 MB buffered. Peak heap lands around 80-100 MB including the merged
+    // stations, which the container absorbs; see the note on the split above
+    // for the part that actually matters. Body.text() decodes UTF-8 and strips
+    // a BOM per spec, so neither needs handling here.
+    const { stations, stats } = parseBnetzaTsv(await res.text());
+
+    console.log(
+      `[${this.source}] DE: ${stats.totalRows} rows → ${stations.length} stations ` +
+        `(merged ${stats.mergedDuplicates}, skipped ${stats.notOperational} not operational, ` +
+        `${stats.outOfBbox} out of bounds, ${stats.malformed} malformed)`,
+    );
+    // ~1% of merges disagree on operator — colocated chargers run by different
+    // companies, which is real and expected. Only warn if that share jumps,
+    // which is what over-merging would actually look like.
+    if (stats.operatorConflicts > stats.mergedDuplicates * OPERATOR_CONFLICT_WARN_RATIO) {
+      console.warn(
+        `[${this.source}] DE: ${stats.operatorConflicts} of ${stats.mergedDuplicates} merged ` +
+          `row(s) disagreed on operator — the merge precision may be too coarse`,
+      );
+    }
+
+    // The register publishes no tariff of any kind, and an EV tariff is per-kWh
+    // and per-session anyway, which fuel_prices cannot express. Nothing is
+    // invented here.
+    return { stations, prices: [] };
+  }
+
+  /**
+   * Run the normal pipeline, then take out the rows nothing else will.
+   *
+   * `base.run()` only orphan-cleans price-less `fuel` stations, so EV rows
+   * accumulate forever. Two sweeps are needed:
+   *
+   *   1. Stations that left the register (or whose derived ID churned because
+   *      an operator corrected its coordinates) — without this the map degrades
+   *      monotonically, one stale pin per correction.
+   *   2. The OpenChargeMap rows this source replaces. Germany is served by
+   *      exactly one EV source (see germany-ev-source.ts), so once BNetzA has
+   *      landed, the `ocm-` rows are pure duplicates. Unlike Spain's REVE
+   *      handover there is no gradual backfill to wait out: this file arrives
+   *      complete in a single run, so one healthy run is the whole cutover.
+   *
+   * Both are gated on a DB-counted floor rather than on the `stationsUpserted`
+   * the pipeline reports, because that figure is `batch.length` regardless of
+   * what the database actually did.
+   */
+  async run(): Promise<ScraperResult> {
+    const result = await super.run();
+    if (result.errors.length > 0) return result;
+
+    try {
+      await this.sweep(result.durationMs);
+    } catch (err) {
+      const msg = err instanceof Error ? err.message : String(err);
+      console.error(`[${this.source}] Cleanup failed: ${msg}`);
+      result.errors.push(`Cleanup: ${msg}`);
+    }
+    return result;
+  }
+
+  private async sweep(runDurationMs: number): Promise<void> {
+    const adapter = new PrismaPg({ connectionString: process.env.DATABASE_URL! });
+    const prisma = new PrismaClient({ adapter });
+    try {
+      const counts: Array<{ count: bigint }> = await prisma.$queryRawUnsafe(
+        `SELECT count(*) FROM stations
+         WHERE country = 'DE' AND station_type = 'ev_charger'
+           AND external_id LIKE 'bnetza-%'`,
+      );
+      const stored = Number(counts[0]?.count ?? 0);
+      if (stored < MIN_STATIONS) {
+        console.warn(
+          `[${this.source}] Only ${stored} station(s) stored (floor ${MIN_STATIONS}) — ` +
+            `skipping cleanup`,
+        );
+        return;
+      }
+
+      // Cutoff expressed as a duration back from the database clock rather than
+      // as an app-side timestamp, so clock skew between app and DB cannot make
+      // this delete rows that were just written.
+      const staleSeconds = runDurationMs / 1000 + SWEEP_MARGIN_SECONDS;
+      const stale: Array<{ count: bigint }> = await prisma.$queryRawUnsafe(
+        `WITH deleted AS (
+           DELETE FROM stations
+           WHERE country = 'DE'
+             AND station_type = 'ev_charger'
+             AND external_id LIKE 'bnetza-%'
+             AND updated_at < NOW() - make_interval(secs => $1::float8)
+           RETURNING id
+         ) SELECT count(*) FROM deleted`,
+        staleSeconds,
+      );
+      const staleCount = Number(stale[0]?.count ?? 0);
+      if (staleCount > 0) {
+        console.log(`[${this.source}] Removed ${staleCount} station(s) no longer in the register`);
+      }
+
+      const retired: Array<{ count: bigint }> = await prisma.$queryRawUnsafe(
+        `WITH deleted AS (
+           DELETE FROM stations
+           WHERE country = 'DE'
+             AND station_type = 'ev_charger'
+             AND external_id LIKE 'ocm-%'
+           RETURNING id
+         ) SELECT count(*) FROM deleted`,
+      );
+      const retiredCount = Number(retired[0]?.count ?? 0);
+      if (retiredCount > 0) {
+        console.log(
+          `[${this.source}] Retired ${retiredCount} superseded OpenChargeMap row(s) for DE ` +
+            `(${stored} registry stations in place)`,
+        );
+      }
+    } finally {
+      await prisma.$disconnect().catch(() => {});
+    }
+  }
+}

From 970b3f88b222f66d7753171ae3e2bf9c49e0f224 Mon Sep 17 00:00:00 2001
From: yellowhat <1692490+yellowhat@users.noreply.github.com>
Date: Thu, 17 Sep 2026 21:05:38 +0200
Subject: [PATCH 3/5] feat(ev): use BNetzA instead of OpenChargeMap for Germany
 by default
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Germany now has two EV sources: Open Charge Map (EV_DE, crowdsourced,
needs an API key) and the BNetzA Ladesäulenregister (EV_DE_BNETZA,
official, keyless, several times larger). They must never both run. The
two overlap heavily, so running both double-pins the country, and once
the BNetzA scraper has retired the ocm- rows a single stray Open Charge
Map run puts every one of them straight back, because nothing
garbage-collects EV rows.

BNetzA wins by default: it is official, complete, and needs no signup.
PUMPERLY_DE_EV_SOURCE=ocm keeps Open Charge Map instead. The rule lives
in one function, germany-ev-source.ts, called by both the scheduler and
the manual CLI, for the same reason as spain-ev-source.ts: two copies
would drift, and the drift is silent.

Unlike Spain's REVE handover there is no backfill period. The register
arrives complete in one daily file, so Open Charge Map stops being
scraped for Germany and its existing German rows are retired on the
first healthy run. Naming EV_DE explicitly on the CLI still runs it.

The legal modal gains the CC BY 4.0 attribution the licence requires.
---
 src/components/nav/legal-modal.tsx     |  1 +
 src/instrumentation.ts                 | 15 +++++++
 src/scrapers/cli.ts                    | 11 +++--
 src/scrapers/germany-ev-source.test.ts | 60 ++++++++++++++++++++++++++
 src/scrapers/germany-ev-source.ts      | 38 ++++++++++++++++
 5 files changed, 122 insertions(+), 3 deletions(-)
 create mode 100644 src/scrapers/germany-ev-source.test.ts
 create mode 100644 src/scrapers/germany-ev-source.ts

diff --git a/src/components/nav/legal-modal.tsx b/src/components/nav/legal-modal.tsx
index a5e89b5..744e384 100644
--- a/src/components/nav/legal-modal.tsx
+++ b/src/components/nav/legal-modal.tsx
@@ -159,6 +159,7 @@ function SourcesContent() {
       <ul>
         <li><strong>Open Charge Map</strong> &mdash; EV charging station locations across all supported countries. Community-maintained, Open Data Commons Open Database License (ODbL). <a href="https://openchargemap.org" target="_blank" rel="noopener noreferrer">openchargemap.org</a></li>
         <li><strong>Mapa REVE</strong> &mdash; EV charging station locations in Spain. Source: Red El&eacute;ctrica de Espa&ntilde;a, S.A.U. Used non-commercially and reproduced without alteration. <a href="https://www.mapareve.es" target="_blank" rel="noopener noreferrer">mapareve.es</a></li>
+        <li><strong>Ladesäulenregister</strong> &mdash; EV charging station locations in Germany, from the register operators must file under &sect;5 Ladesäulenverordnung. Source: Bundesnetzagentur.de, Creative Commons Attribution 4.0 (CC BY 4.0). <a href="https://www.bundesnetzagentur.de/DE/Fachthemen/ElektrizitaetundGas/E-Mobilitaet/Ladesaeulenkarte/start.html" target="_blank" rel="noopener noreferrer">bundesnetzagentur.de</a></li>
       </ul>
 
       <h3>Map and routing</h3>
diff --git a/src/instrumentation.ts b/src/instrumentation.ts
index f0ff32a..d850897 100644
--- a/src/instrumentation.ts
+++ b/src/instrumentation.ts
@@ -52,6 +52,8 @@ const DEFAULT_INTERVALS: Record<string, number> = {
   // Spain Mapa REVE — the API allows only 5 requests/hour, so this crawls a
   // few pages at a time and must run hourly to get through the registry.
   EV_ES_REVE: 1,
+  // Germany BNetzA Ladesäulenregister — one bulk TSV, regenerated daily.
+  EV_DE_BNETZA: 24,
 };
 
 export async function register() {
@@ -105,6 +107,8 @@ export async function register() {
   const { OCMScraper } = await import("./scrapers/ocm");
   const { REVEScraper } = await import("./scrapers/reve");
   const { resolveSpainEvSource } = await import("./scrapers/spain-ev-source");
+  const { BNetzAScraper } = await import("./scrapers/bnetza");
+  const { resolveGermanyEvSource } = await import("./scrapers/germany-ev-source");
   const { StaticScraper } = await import("./scrapers/static");
   const { STATIC_DATASETS } = await import("./scrapers/data");
 
@@ -187,6 +191,8 @@ export async function register() {
     EV_US: () => new OCMScraper("US"),
     // Spain's official EV registry — supersedes EV_ES when a key is set (#121)
     EV_ES_REVE: () => new REVEScraper(),
+    // Germany's official EV registry — supersedes EV_DE by default, keyless
+    EV_DE_BNETZA: () => new BNetzAScraper(),
   };
 
   // Register community-contributed static datasets (see scrapers/data/README.md).
@@ -239,6 +245,15 @@ export async function register() {
     );
   }
 
+  // Germany likewise — see scrapers/germany-ev-source.ts, and scrapers/bnetza.ts
+  // for the sweep that retires the OpenChargeMap rows once BNetzA has landed.
+  countries = resolveGermanyEvSource(countries);
+  if (countries.includes("EV_DE_BNETZA")) {
+    console.log(
+      "[scraper] Germany EV: using the BNetzA Ladesäulenregister instead of OpenChargeMap",
+    );
+  }
+
   // Resolve per-country intervals
   for (const code of countries) {
     // Priority: PUMPERLY_SCRAPE_INTERVAL_XX > PUMPERLY_SCRAPE_INTERVAL_HOURS > DEFAULT_INTERVALS
diff --git a/src/scrapers/cli.ts b/src/scrapers/cli.ts
index ccccfb7..c189282 100644
--- a/src/scrapers/cli.ts
+++ b/src/scrapers/cli.ts
@@ -40,6 +40,8 @@ import { MexicoScraper } from "./mexico";
 import { OCMScraper } from "./ocm";
 import { REVEScraper } from "./reve";
 import { resolveSpainEvSource } from "./spain-ev-source";
+import { BNetzAScraper } from "./bnetza";
+import { resolveGermanyEvSource } from "./germany-ev-source";
 
 // ---------------------------------------------------------------------------
 // Scraper CLI
@@ -132,6 +134,8 @@ const SCRAPERS: Record<string, Array<() => BaseScraper>> = {
   EV_US: [() => new OCMScraper("US")],
   // Spain's official EV registry — supersedes EV_ES when a key is set (#121)
   EV_ES_REVE: [() => new REVEScraper()],
+  // Germany's official EV registry — supersedes EV_DE by default, keyless
+  EV_DE_BNETZA: [() => new BNetzAScraper()],
 };
 
 function usage(): never {
@@ -154,11 +158,12 @@ function parseArgs(argv: string[]): { countries: string[] } {
     usage();
   }
 
-  // `all` must not run both Spanish EV sources. Naming EV_ES explicitly still
-  // works — the CLI is a manual override, and asking for it by name means it.
+  // `all` must not run both Spanish or both German EV sources. Naming EV_ES or
+  // EV_DE explicitly still works — the CLI is a manual override, and asking for
+  // it by name means it.
   const countries =
     countryArg === "ALL"
-      ? resolveSpainEvSource(Object.keys(SCRAPERS))
+      ? resolveGermanyEvSource(resolveSpainEvSource(Object.keys(SCRAPERS)))
       : countryArg.split(",").map((c) => c.trim().toUpperCase());
 
   // Validate
diff --git a/src/scrapers/germany-ev-source.test.ts b/src/scrapers/germany-ev-source.test.ts
new file mode 100644
index 0000000..498ab09
--- /dev/null
+++ b/src/scrapers/germany-ev-source.test.ts
@@ -0,0 +1,60 @@
+import { describe, it, expect, afterEach, vi } from "vitest";
+import { resolveGermanyEvSource } from "./germany-ev-source";
+
+// The scheduler and the manual CLI both route Germany's EV scraping through
+// this, so this is the test that stops them drifting apart. Drift is silent:
+// the map just quietly refills with duplicate German chargers.
+
+describe("resolveGermanyEvSource", () => {
+  afterEach(() => {
+    vi.unstubAllEnvs();
+  });
+
+  // What the scheduler produces once EV_XX codes are derived.
+  const SCHEDULED = ["DE", "FR", "EV_DE", "EV_FR", "EV_US"];
+  // What `--country=all` produces: both German sources are registered.
+  const ALL = ["DE", "FR", "EV_DE", "EV_FR", "EV_DE_BNETZA", "EV_US"];
+
+  it("uses BNetzA for Germany by default", () => {
+    expect(resolveGermanyEvSource(SCHEDULED)).toContain("EV_DE_BNETZA");
+    expect(resolveGermanyEvSource(SCHEDULED)).not.toContain("EV_DE");
+  });
+
+  it("keeps OpenChargeMap when PUMPERLY_DE_EV_SOURCE=ocm", () => {
+    vi.stubEnv("PUMPERLY_DE_EV_SOURCE", "ocm");
+    expect(resolveGermanyEvSource(SCHEDULED)).toContain("EV_DE");
+    expect(resolveGermanyEvSource(SCHEDULED)).not.toContain("EV_DE_BNETZA");
+  });
+
+  it("is case- and whitespace-insensitive about the opt-out", () => {
+    vi.stubEnv("PUMPERLY_DE_EV_SOURCE", "  OCM  ");
+    expect(resolveGermanyEvSource(SCHEDULED)).toContain("EV_DE");
+  });
+
+  it("falls back to BNetzA when the override is not a source it knows", () => {
+    vi.stubEnv("PUMPERLY_DE_EV_SOURCE", "nonsense");
+    expect(resolveGermanyEvSource(SCHEDULED)).toContain("EV_DE_BNETZA");
+  });
+
+  it("collapses --country=all down to one German EV source", () => {
+    for (const source of ["ocm", ""]) {
+      vi.stubEnv("PUMPERLY_DE_EV_SOURCE", source);
+      const out = resolveGermanyEvSource(ALL);
+      expect(out.filter((c) => c === "EV_DE" || c === "EV_DE_BNETZA")).toHaveLength(1);
+    }
+  });
+
+  it("leaves every other country untouched", () => {
+    expect(resolveGermanyEvSource(ALL).filter((c) => !c.startsWith("EV_DE"))).toEqual([
+      "DE",
+      "FR",
+      "EV_FR",
+      "EV_US",
+    ]);
+  });
+
+  it("adds nothing when Germany has no EV scraper enabled", () => {
+    const fuelOnly = ["DE", "FR", "IT"];
+    expect(resolveGermanyEvSource(fuelOnly)).toEqual(fuelOnly);
+  });
+});
diff --git a/src/scrapers/germany-ev-source.ts b/src/scrapers/germany-ev-source.ts
new file mode 100644
index 0000000..a5b7b49
--- /dev/null
+++ b/src/scrapers/germany-ev-source.ts
@@ -0,0 +1,38 @@
+// ---------------------------------------------------------------------------
+// Which source supplies Germany's EV chargers
+// ---------------------------------------------------------------------------
+// Germany has two: OpenChargeMap (`EV_DE`, crowdsourced, needs an API key) and
+// the BNetzA Ladesäulenregister (`EV_DE_BNETZA`, the official register every
+// operator of a public charge point must file into, keyless).
+//
+// They must never both run. The register holds ~74k German locations against
+// OpenChargeMap's crowdsourced subset, and the two overlap heavily — running
+// both double-pins the country. Worse, once the BNetzA scraper has retired the
+// `ocm-` rows (see scrapers/bnetza.ts), a single stray OpenChargeMap run puts
+// every one of them straight back, and nothing garbage-collects EV rows.
+//
+// BNetzA wins by default: it is official, it needs no key, and it is the more
+// complete of the two. `PUMPERLY_DE_EV_SOURCE=ocm` is the escape hatch.
+//
+// This rule is needed by both the scheduler (instrumentation.ts) and the manual
+// CLI (scrapers/cli.ts), so it lives here rather than in either of them — the
+// same reasoning as spain-ev-source.ts, whose shape this mirrors exactly. Two
+// copies would drift, and the way they drift is silent: the map just quietly
+// fills up with duplicates again.
+// ---------------------------------------------------------------------------
+
+/**
+ * Collapse Germany's two EV scrapers down to whichever one is configured.
+ *
+ * Returns `codes` unchanged when Germany has no EV scraper enabled at all (e.g.
+ * PUMPERLY_EV_ENABLED=0). Otherwise exactly one of `EV_DE` / `EV_DE_BNETZA`
+ * survives: OpenChargeMap when PUMPERLY_DE_EV_SOURCE=ocm, BNetzA otherwise.
+ */
+export function resolveGermanyEvSource(codes: string[]): string[] {
+  const hasGermanyEv = codes.includes("EV_DE") || codes.includes("EV_DE_BNETZA");
+  const rest = codes.filter((c) => c !== "EV_DE" && c !== "EV_DE_BNETZA");
+  if (!hasGermanyEv) return rest;
+  const preferOcm = process.env.PUMPERLY_DE_EV_SOURCE?.trim().toLowerCase() === "ocm";
+  rest.push(preferOcm ? "EV_DE" : "EV_DE_BNETZA");
+  return rest;
+}

From bb6a6d262998a9fa5b10bd4bd66dabb8315aa8d2 Mon Sep 17 00:00:00 2001
From: yellowhat <1692490+yellowhat@users.noreply.github.com>
Date: Thu, 17 Sep 2026 21:06:43 +0200
Subject: [PATCH 4/5] docs(ev): document the Germany BNetzA source and its env
 vars
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

README: list the Ladesäulenregister alongside Open Charge Map and Mapa
REVE in the features and EV sources table (CC BY 4.0, attribution
"Bundesnetzagentur.de"), explain the default switch for Germany and the
PUMPERLY_DE_EV_SOURCE=ocm opt-out, and add PUMPERLY_DE_EV_SOURCE and
PUMPERLY_BNETZA_MIN_STATIONS to the environment variable table.

.env.example: the same two variables with their defaults and the reason
the safety floor exists.
---
 .env.example | 13 +++++++++++++
 README.md    |  7 ++++++-
 2 files changed, 19 insertions(+), 1 deletion(-)

diff --git a/.env.example b/.env.example
index ae0faf5..87b5fba 100644
--- a/.env.example
+++ b/.env.example
@@ -89,3 +89,16 @@ PUMPERLY_DEFAULT_COUNTRY=ES
 
 # Set to 0 to disable EV charger scraping (default: enabled when OCM key is set)
 # PUMPERLY_EV_ENABLED=1
+
+# Germany EV chargers: which of the two sources supplies them. Defaults to the
+# BNetzA Ladesäulenregister — the official register every operator of a public
+# charge point must file into, ~74,000 locations, no API key and no signup.
+# Set to "ocm" to keep OpenChargeMap instead. The two never run together: the
+# BNetzA scraper retires OpenChargeMap's German rows on its first healthy run,
+# and running both would double-pin the country.
+# PUMPERLY_DE_EV_SOURCE=bnetza
+
+# Stations that must be stored before the BNetzA scraper is allowed to prune
+# stale rows or retire OpenChargeMap's German ones. A safety floor: a degraded
+# fetch must not be able to delete a working map.
+# PUMPERLY_BNETZA_MIN_STATIONS=10000
diff --git a/README.md b/README.md
index 5b098dc..e4f7bab 100644
--- a/README.md
+++ b/README.md
@@ -48,7 +48,7 @@ Pumperly combines route planning with real-time fuel prices and EV charging stat
 
 - **Route planning** — Geocoding via [Photon](https://github.com/komoot/photon), routing via [Valhalla](https://github.com/valhalla/valhalla), with alternative routes
 - **Real-time fuel prices** — From government open data APIs and community sources
-- **EV charging stations** — Via [Open Charge Map](https://openchargemap.org) across all supported countries, and the official [Mapa REVE](https://www.mapareve.es) registry in Spain
+- **EV charging stations** — Via [Open Charge Map](https://openchargemap.org) across all supported countries, plus the official [Mapa REVE](https://www.mapareve.es) registry in Spain and the [BNetzA Ladesäulenregister](https://www.bundesnetzagentur.de/DE/Fachthemen/ElektrizitaetundGas/E-Mobilitaet/Ladesaeulenkarte/start.html) in Germany
 - **Detour calculation** — Each station shows estimated detour time from your route
 - **"Cheapest within N min"** — Slider filters stations by maximum detour, highlights the best deal
 - **Corridor station list** — Sorted by position along route, with price deltas vs average
@@ -98,9 +98,12 @@ Pumperly combines route planning with real-time fuel prices and EV charging stat
 |---|---|---|
 | [Open Charge Map](https://openchargemap.org) | All supported countries + United States (EV-only) | ODbL |
 | [Mapa REVE](https://www.mapareve.es) (Red Eléctrica de España) | Spain — official operator-reported registry | Non-commercial, attribution required |
+| [Ladesäulenregister](https://www.bundesnetzagentur.de/DE/Fachthemen/ElektrizitaetundGas/E-Mobilitaet/Ladesaeulenkarte/start.html) (Bundesnetzagentur) | Germany — official operator-reported registry, ~74,000 locations | CC BY 4.0, attribution "Bundesnetzagentur.de" |
 
 Spain uses Mapa REVE when `PUMPERLY_REVE_API_KEY` is set: it is the registry every Spanish charge point operator files into, so it is authoritative where Open Charge Map is crowdsourced. The two overlap heavily, so Open Charge Map stops being scraped for Spain immediately, its existing Spanish rows stay visible while REVE backfills, and they are deleted once REVE reaches 95% of the registry. Expect duplicate Spanish pins until then.
 
+Germany uses the BNetzA Ladesäulenregister by default — no API key, no signup. Every operator of a publicly accessible charge point must file into it under §5 Ladesäulenverordnung, so like REVE it is authoritative where Open Charge Map is crowdsourced, and it is several times larger. It arrives as one daily bulk file, so there is no backfill period: Open Charge Map stops being scraped for Germany, and its existing German rows are retired on the first healthy run. Set `PUMPERLY_DE_EV_SOURCE=ocm` to keep Open Charge Map instead.
+
 ### Map & routing
 
 | Service | Purpose | License |
@@ -413,6 +416,8 @@ volumes:
 | `PUMPERLY_PRICE_MIN` / `PUMPERLY_PRICE_MAX` | Price bounds for scraper validation (EUR/L) | `0.30` / `4.00` |
 | `PUMPERLY_SCRAPE_INTERVAL_HOURS` | Global scrape interval override (hours, 0=disable) | Per-country |
 | `PUMPERLY_EV_ENABLED` | Enable EV charger scraping (`0` to disable) | `1` |
+| `PUMPERLY_DE_EV_SOURCE` | Germany's EV source: `bnetza` (official registry) or `ocm` | `bnetza` |
+| `PUMPERLY_BNETZA_MIN_STATIONS` | Stations that must be stored before the BNetzA scraper prunes stale rows or retires Open Charge Map's German ones | `10000` |
 | `VALHALLA_URL` | Valhalla routing endpoint | — |
 | `PHOTON_URL` | Photon geocoding endpoint | — |
 

From 25ba9da8abfeeb070ab609678f132f72e9015ed3 Mon Sep 17 00:00:00 2001
From: yellowhat <1692490+yellowhat@users.noreply.github.com>
Date: Thu, 17 Sep 2026 21:23:54 +0200
Subject: [PATCH 5/5] fix(ev): gate the BNetzA cleanup on rows refreshed by the
 current run

The safety floor counted every bnetza- row in the table, old and new alike.
With 74,000 rows from yesterday and a degraded fetch that refreshed only 42,
the count still cleared PUMPERLY_BNETZA_MIN_STATIONS, so the stale sweep
deleted the 74,000 and the retirement sweep dropped the OpenChargeMap rows,
leaving Germany with the degraded dataset.

The count now carries the same updated_at cutoff the stale sweep uses, so
only rows written by this run can vouch for it and every row sits on exactly
one side of the cutoff. Adds a regression test with 74,000 stale rows and 42
refreshed ones that fails against the previous query.
---
 .env.example                |  2 +-
 README.md                   |  2 +-
 src/scrapers/bnetza.test.ts | 38 +++++++++++++++++++++++++++++++++++--
 src/scrapers/bnetza.ts      | 30 ++++++++++++++++++-----------
 4 files changed, 57 insertions(+), 15 deletions(-)

diff --git a/.env.example b/.env.example
index 87b5fba..c49b3f4 100644
--- a/.env.example
+++ b/.env.example
@@ -98,7 +98,7 @@ PUMPERLY_DEFAULT_COUNTRY=ES
 # and running both would double-pin the country.
 # PUMPERLY_DE_EV_SOURCE=bnetza
 
-# Stations that must be stored before the BNetzA scraper is allowed to prune
+# Stations a run must refresh before the BNetzA scraper is allowed to prune
 # stale rows or retire OpenChargeMap's German ones. A safety floor: a degraded
 # fetch must not be able to delete a working map.
 # PUMPERLY_BNETZA_MIN_STATIONS=10000
diff --git a/README.md b/README.md
index e4f7bab..91bf3a1 100644
--- a/README.md
+++ b/README.md
@@ -417,7 +417,7 @@ volumes:
 | `PUMPERLY_SCRAPE_INTERVAL_HOURS` | Global scrape interval override (hours, 0=disable) | Per-country |
 | `PUMPERLY_EV_ENABLED` | Enable EV charger scraping (`0` to disable) | `1` |
 | `PUMPERLY_DE_EV_SOURCE` | Germany's EV source: `bnetza` (official registry) or `ocm` | `bnetza` |
-| `PUMPERLY_BNETZA_MIN_STATIONS` | Stations that must be stored before the BNetzA scraper prunes stale rows or retires Open Charge Map's German ones | `10000` |
+| `PUMPERLY_BNETZA_MIN_STATIONS` | Stations a run must refresh before the BNetzA scraper prunes stale rows or retires Open Charge Map's German ones | `10000` |
 | `VALHALLA_URL` | Valhalla routing endpoint | — |
 | `PHOTON_URL` | Photon geocoding endpoint | — |
 
diff --git a/src/scrapers/bnetza.test.ts b/src/scrapers/bnetza.test.ts
index b648714..aac7daf 100644
--- a/src/scrapers/bnetza.test.ts
+++ b/src/scrapers/bnetza.test.ts
@@ -413,12 +413,13 @@ describe("BNetzAScraper cleanup", () => {
 
   async function setup(opts: {
     results?: Array<Array<{ count: bigint }>>;
+    client?: ReturnType<typeof stubPrisma>;
     errors?: string[];
     durationMs?: number;
   }) {
     const { PrismaClient } = await import("../generated/prisma/client");
     const { BaseScraper } = await import("./base");
-    const stub = stubPrisma(opts.results ?? []);
+    const stub = opts.client ?? stubPrisma(opts.results ?? []);
     // A plain function, not an arrow: this stands in for a constructor.
     vi.mocked(PrismaClient).mockImplementation(function () {
       return stub.client;
@@ -444,7 +445,13 @@ describe("BNetzAScraper cleanup", () => {
 
     expect(result.errors).toEqual([]);
     expect(stub.queries).toHaveLength(3);
+
+    // The floor counts only rows this run refreshed, using the same cutoff as
+    // the stale sweep, so rows left over from an earlier run cannot clear it.
     expect(stub.queries[0].sql).toMatch(/count\(\*\)/);
+    expect(stub.queries[0].sql).toMatch(/external_id LIKE 'bnetza-%'/);
+    expect(stub.queries[0].sql).toMatch(/updated_at >= NOW\(\) - make_interval/);
+    expect(stub.queries[0].params[0]).toBe(30 + 300);
 
     // Stale sweep: scoped to this source, and cut off by a duration measured
     // back from the DB clock so app/DB clock skew cannot delete fresh rows.
@@ -458,7 +465,34 @@ describe("BNetzAScraper cleanup", () => {
     expect(stub.queries[2].sql).toMatch(/station_type = 'ev_charger'/);
   });
 
-  it("deletes nothing when too few stations were stored", async () => {
+  it("leaves yesterday's rows alone when a degraded fetch refreshed too few of them", async () => {
+    // 74,000 rows from yesterday's run plus 42 from today's degraded fetch.
+    // Counting them together would clear the floor and delete the 74,000.
+    const rowAgesSeconds = [...Array<number>(74_000).fill(86_400), ...Array<number>(42).fill(5)];
+    const queries: Array<{ sql: string; params: unknown[] }> = [];
+    const client = {
+      $queryRawUnsafe: vi.fn(async (sql: string, ...params: unknown[]) => {
+        queries.push({ sql, params });
+        if (/DELETE/.test(sql)) throw new Error("sweep ran against a degraded fetch");
+        // Answer the count the way Postgres would: apply the freshness window
+        // if the statement has one, otherwise count every row.
+        const window = /updated_at >= NOW\(\) - make_interval/.test(sql)
+          ? Number(params[0])
+          : Infinity;
+        return [{ count: BigInt(rowAgesSeconds.filter((age) => age <= window).length) }];
+      }),
+      $disconnect: vi.fn(async () => {}),
+    };
+    await setup({ client: { queries, client } });
+    const { BNetzAScraper } = await import("./bnetza");
+
+    const result = await new BNetzAScraper().run();
+
+    expect(result.errors).toEqual([]);
+    expect(queries).toHaveLength(1);
+  });
+
+  it("deletes nothing when too few stations were refreshed", async () => {
     // A degraded fetch that still clears base.run()'s empty-fetch guard must
     // not be able to wipe yesterday's map.
     const stub = await setup({ results: [[{ count: BigInt(42) }]] });
diff --git a/src/scrapers/bnetza.ts b/src/scrapers/bnetza.ts
index 621b33c..e38bd00 100644
--- a/src/scrapers/bnetza.ts
+++ b/src/scrapers/bnetza.ts
@@ -356,7 +356,10 @@ export class BNetzAScraper extends BaseScraper {
    *
    * Both are gated on a DB-counted floor rather than on the `stationsUpserted`
    * the pipeline reports, because that figure is `batch.length` regardless of
-   * what the database actually did.
+   * what the database actually did. The count covers only rows this run
+   * touched: yesterday's 74k rows must not vouch for a degraded fetch that
+   * refreshed 42 of them, or sweep 1 would delete the other 73,958 and sweep 2
+   * would retire OpenChargeMap on the strength of a near-empty file.
    */
   async run(): Promise<ScraperResult> {
     const result = await super.run();
@@ -376,24 +379,29 @@ export class BNetzAScraper extends BaseScraper {
     const adapter = new PrismaPg({ connectionString: process.env.DATABASE_URL! });
     const prisma = new PrismaClient({ adapter });
     try {
+      // Cutoff expressed as a duration back from the database clock rather than
+      // as an app-side timestamp, so clock skew between app and DB cannot make
+      // this delete rows that were just written. The same cutoff separates
+      // "refreshed by this run" from "stale" below, so a row is always on
+      // exactly one side of it.
+      const staleSeconds = runDurationMs / 1000 + SWEEP_MARGIN_SECONDS;
+
       const counts: Array<{ count: bigint }> = await prisma.$queryRawUnsafe(
         `SELECT count(*) FROM stations
          WHERE country = 'DE' AND station_type = 'ev_charger'
-           AND external_id LIKE 'bnetza-%'`,
+           AND external_id LIKE 'bnetza-%'
+           AND updated_at >= NOW() - make_interval(secs => $1::float8)`,
+        staleSeconds,
       );
-      const stored = Number(counts[0]?.count ?? 0);
-      if (stored < MIN_STATIONS) {
+      const refreshed = Number(counts[0]?.count ?? 0);
+      if (refreshed < MIN_STATIONS) {
         console.warn(
-          `[${this.source}] Only ${stored} station(s) stored (floor ${MIN_STATIONS}) — ` +
-            `skipping cleanup`,
+          `[${this.source}] Only ${refreshed} station(s) refreshed by this run ` +
+            `(floor ${MIN_STATIONS}) — skipping cleanup`,
         );
         return;
       }
 
-      // Cutoff expressed as a duration back from the database clock rather than
-      // as an app-side timestamp, so clock skew between app and DB cannot make
-      // this delete rows that were just written.
-      const staleSeconds = runDurationMs / 1000 + SWEEP_MARGIN_SECONDS;
       const stale: Array<{ count: bigint }> = await prisma.$queryRawUnsafe(
         `WITH deleted AS (
            DELETE FROM stations
@@ -423,7 +431,7 @@ export class BNetzAScraper extends BaseScraper {
       if (retiredCount > 0) {
         console.log(
           `[${this.source}] Retired ${retiredCount} superseded OpenChargeMap row(s) for DE ` +
-            `(${stored} registry stations in place)`,
+            `(${refreshed} registry stations refreshed this run)`,
         );
       }
     } finally {