// Line Bites OpenStreetMap refresh: the food & drink places around each city's stations, rebuilt
// weekly from Geofabrik's regional extracts. Traps: CLAUDE.md next to this file. Design,
// measurements and ODbL notes: NOTES.md.
//
// This file is also the published extraction method for the ODbL share-alike obligation (served
// at https://linebites.bence.dev/odbl/refresh-osm.mjs).
//
//   node refresh-osm.mjs [cityId ...]      (default: every city in stations.json)
//
// First every <region>-latest.osm.pbf the target cities need is checked with a HEAD: a region
// Geofabrik renamed or retired fails its cities before any of their other extracts is downloaded.
// Cities are then taken stalest first (never built, then oldest generatedAt), and a city is only
// started when all its extracts can still be fetched before the deadline. Every Geofabrik extract
// that a city needs (EXTRACTS below) is handled once per run, one at a time:
//   1. download the dated .osm.pbf that <region>-latest.osm.pbf points to, checked against its .md5;
//   2. with osmium, take the objects that have a tag the Line Bites app maps (line-bites
//      Sources/Places/OverpassQuery.swift) plus the nodes and member ways their shapes need, and
//      `osmium export` them as GeoJSON lines (createOsmium().features: how, and why not in one
//      `osmium tags-filter`); then delete the download;
//   3. each named, categorised object becomes a place at the node, or at the centroid of the way's
//      or multipolygon's outer ring(s), and every city that uses the extract keeps the places within
//      1.5 km of one of its stations.
// When the last of a city's extracts is in, <cityId>.json (+ .gz) is written atomically and
// index.json is updated. A city whose extract failed keeps its previous file.
//
// Env: DATA_DIR (/data: the served files), WORK_DIR (/work: downloads, never the served volume),
//      STATIONS (./stations.json), OSMIUM (osmium), DEADLINE_SECONDS (16200),
//      ALLOW_SHRINK=1 (accept a city whose place count more than halved),
//      PUSHGATEWAY_URL (unset: no metrics; the run's outcome and each city's age, see metricsText).
//      City ids as arguments: only those cities, and only the extracts they need.
//
// Exit status: 0 fine, 1 an extract or a city failed (its previous file stays), or two extracts in a
// row failed to transfer and the run stopped, 75 Geofabrik refused us (403/429) and the run stopped,
// 143 SIGTERM. The CronJob never retries.
//
// Zero npm dependencies: the pod runs this straight out of a ConfigMap on the stock node image;
// osmium-tool comes from Debian, installed by the pod's init container.

import { spawn } from 'node:child_process'
import { createHash, randomUUID } from 'node:crypto'
import { realpathSync, writeSync } from 'node:fs'
import { mkdir, open, readdir, readFile, rename, rm, stat, unlink, utimes } from 'node:fs/promises'
import { hostname } from 'node:os'
import { join } from 'node:path'
import { StringDecoder } from 'node:string_decoder'
import { setTimeout as delay } from 'node:timers/promises'
import { fileURLToPath } from 'node:url'
import { gzipSync } from 'node:zlib'

// ----------------------------------------------------------------------------------------- contract

export const GEOFABRIK_URL = 'https://download.geofabrik.de'
// Says who we are and where the operator is. Geofabrik publishes no formal limits; this client
// downloads one file at a time, each at most once per run, once a week (NOTES.md).
export const USER_AGENT = 'LineBites-cache/1.0 (+https://linebites.bence.dev/odbl)'
export const RADIUS_M = 1500
// The work emptyDir's sizeLimit is 3Gi; the largest extract in EXTRACTS (quebec) was 1.16 GB on
// 2026-09-25. A bigger one is refused before it can get the pod evicted.
export const MAX_EXTRACT_BYTES = 2_500_000_000
// A broken download (network error, 5xx, short body, stall, too slow, md5 mismatch) is tried once
// more after this, resuming with a Range request from the bytes it already has (not after an md5
// mismatch: then the file is wrong somewhere, and it starts over).
export const RETRY_WAIT_MS = 120_000
// A download that delivers nothing for this long is cut off.
export const STALL_MS = 120_000
// A download that averages less than MIN_BYTES_PER_SEC over a SLOW_WINDOW_MS window is cut off too:
// a trickle refreshes the stall timer forever. 100 kB/s would still bring quebec in 3.2 h.
export const SLOW_WINDOW_MS = 10 * 60_000
export const MIN_BYTES_PER_SEC = 100_000
// No extract is started with less than this left before the deadline, and no city with less than
// this per extract it still needs.
export const MIN_EXTRACT_MS = 15 * 60_000
// A transfer still running this long after the deadline is abandoned, and so is one that at its
// current rate would not end by then: the pod's hard stop (activeDeadlineSeconds) is later still,
// with time for osmium (gen.test.mjs checks the sums).
export const TRANSFER_GRACE_MS = 15 * 60_000
// Two extracts in a row that fail to transfer (after their one retry) stop the run: the link or
// Geofabrik is in trouble, and every further extract would be the same waste.
export const TRANSFER_BREAKER = 2
export const EXIT = Object.freeze({ ok: 0, failed: 1, refused: 75, signal: 143 })
export const LICENSE = Object.freeze({
  license: 'ODbL-1.0',
  licenseUrl: 'https://opendatacommons.org/licenses/odbl/1-0/',
  attribution: '© OpenStreetMap contributors',
  source: 'https://www.openstreetmap.org/copyright',
})

// City id -> the Geofabrik extracts (paths under GEOFABRIK_URL, without "-latest.osm.pbf") that
// together contain every point within RADIUS_M of the city's stations. refresh-osm.test.mjs checks
// that against the extracts' clipping polygons, recorded in geofabrik-regions.json by gen-regions.py.
// An extract that several cities share is downloaded once per run. A city missing here fails.
// budapest, brno, prague and tokyo are not in stations.json yet; their entries are ready for it.
export const EXTRACTS = deepFreeze({
  sf: ['north-america/us/california/norcal'],
  nyc: ['north-america/us/new-york', 'north-america/us/new-jersey'],
  boston: ['north-america/us/massachusetts'],
  chicago: ['north-america/us/illinois'],
  la: ['north-america/us/california/socal'],
  montreal: ['north-america/canada/quebec'],
  // RER D runs out of Ile-de-France: Creil, Chantilly and Orry-la-Ville (Oise) and Malesherbes (Loiret).
  paris: ['europe/france/ile-de-france', 'europe/france/picardie', 'europe/france/centre'],
  philadelphia: ['north-america/us/pennsylvania', 'north-america/us/new-jersey'],
  // The airport station's 1.5 km reaches the Columbia, into washington's clipping polygon, but
  // oregon's (buffered past the state line) contains those points too: washington adds nothing.
  portland: ['north-america/us/oregon'],
  seattle: ['north-america/us/washington'],
  toronto: ['north-america/canada/ontario'],
  dc: ['north-america/us/district-of-columbia', 'north-america/us/maryland', 'north-america/us/virginia'],
  budapest: ['europe/hungary'],
  // The Czech regions, not the 0.95 GB country: 0.27 GB for both cities. Stredocesky's polygon
  // includes Prague (it has no hole), so praha itself is not needed.
  brno: ['europe/czech-republic/jihomoravsky'],
  prague: ['europe/czech-republic/stredocesky'],
  tokyo: ['asia/japan/kanto'],
})

function deepFreeze(table) {
  for (const list of Object.values(table)) Object.freeze(list)
  return Object.freeze(table)
}

// The app's OSM mapping, in its PlaceCategory order. Keep in lockstep with OverpassQuery.swift:
// refresh-osm.test.mjs compares these tables with the Swift source when the app repo is checked out.
export const AMENITY = Object.freeze({
  restaurant: ['restaurant', 'food_court'],
  fastFood: ['fast_food'],
  cafe: ['cafe'],
  bakery: ['ice_cream'],
  bar: ['bar', 'pub', 'biergarten', 'nightclub'],
  market: ['marketplace'],
})
export const SHOP = Object.freeze({
  bakery: ['bakery', 'pastry', 'confectionery', 'chocolate'],
  bottleShop: ['alcohol', 'wine', 'beverages'],
  market: ['deli', 'greengrocer', 'cheese', 'supermarket', 'coffee', 'tea'],
})
export const CRAFT = Object.freeze({
  brewery: ['brewery', 'winery', 'distillery'],
})

const lookup = (table) => new Map(Object.entries(table).flatMap(([category, values]) => values.map((v) => [v, category])))
const AMENITY_TO = lookup(AMENITY)
const SHOP_TO = lookup(SHOP)
const CRAFT_TO = lookup(CRAFT)

/** Same precedence as the app: craft, then a microbrewery bar, then amenity, then shop. */
export function categoryFor(tags) {
  if (!tags) return null
  if (CRAFT_TO.has(tags.craft)) return CRAFT_TO.get(tags.craft)
  if (tags.amenity === 'bar' && tags.microbrewery === 'yes') return 'brewery'
  if (AMENITY_TO.has(tags.amenity)) return AMENITY_TO.get(tags.amenity)
  if (SHOP_TO.has(tags.shop)) return SHOP_TO.get(tags.shop)
  return null
}

/**
 * `osmium tags-filter` expressions for exactly the values the tables map: nodes, ways and relations
 * with one of them. amenity=bar covers the microbrewery rule. `name` is checked afterwards, here:
 * one expression cannot require two tags.
 */
export function tagFilters() {
  return [
    ['amenity', AMENITY],
    ['shop', SHOP],
    ['craft', CRAFT],
  ].map(([key, table]) => `nwr/${key}=${Object.values(table).flat().join(',')}`)
}

// ------------------------------------------------------------------------------------------ geometry

const EARTH_M = 6371008.8
const rad = (deg) => (deg * Math.PI) / 180
// Metres per degree of latitude is 110,574 at the equator and more elsewhere: the smallest value
// gives the widest (safe) grid cell. Longitude uses the cosine of the most poleward station's cell.
const M_PER_DEG_LAT_MIN = 110_574
const M_PER_DEG_LON_EQ = 111_320

export function distanceM(lat1, lon1, lat2, lon2) {
  const a = Math.sin(rad(lat2 - lat1) / 2) ** 2 + Math.cos(rad(lat1)) * Math.cos(rad(lat2)) * Math.sin(rad(lon2 - lon1) / 2) ** 2
  return 2 * EARTH_M * Math.asin(Math.min(1, Math.sqrt(a)))
}

export const round6 = (x) => Math.round(x * 1e6) / 1e6

/** "Is this point within radiusM of any station?", via a grid so a city of 2,000 stations stays cheap. */
export function stationIndex(stations, radiusM = RADIUS_M) {
  const maxAbsLat = Math.max(...stations.map(([lat]) => Math.abs(lat)))
  const cellLat = (radiusM / M_PER_DEG_LAT_MIN) * 1.01
  const cellLon = (radiusM / (M_PER_DEG_LON_EQ * Math.cos(rad(Math.min(89, maxAbsLat + 2 * cellLat))))) * 1.01
  const grid = new Map()
  const key = (i, j) => `${i},${j}`
  for (const [lat, lon] of stations) {
    const k = key(Math.floor(lat / cellLat), Math.floor(lon / cellLon))
    if (!grid.has(k)) grid.set(k, [])
    grid.get(k).push([lat, lon])
  }
  return {
    within(lat, lon) {
      const i = Math.floor(lat / cellLat)
      const j = Math.floor(lon / cellLon)
      for (let di = -1; di <= 1; di++) {
        for (let dj = -1; dj <= 1; dj++) {
          for (const [sLat, sLon] of grid.get(key(i + di, j + dj)) ?? []) {
            if (distanceM(lat, lon, sLat, sLon) <= radiusM) return true
          }
        }
      }
      return false
    },
  }
}

// Centroids are computed on plain [lon, lat] degrees. That is exact for the local projection
// (x = lon * cos(lat0), y = lat): it is an affine map, and centroids commute with affine maps. The
// coordinates are taken relative to the first vertex, so a 10 m outline far from 0,0 keeps its digits.

/**
 * Area centroid of a closed ring and its (unsigned) area in square degrees; null for a ring without
 * area, and for one whose centroid falls outside its own bounding box (see below).
 */
export function ringCentroid(ring) {
  if (!Array.isArray(ring) || ring.length < 4) return null
  const [x0, y0] = ring[0]
  let twiceArea = 0
  let cx = 0
  let cy = 0
  let span = 0
  let minX = 0
  let maxX = 0
  let minY = 0
  let maxY = 0
  for (let i = 0; i < ring.length - 1; i++) {
    const x1 = ring[i][0] - x0
    const y1 = ring[i][1] - y0
    const x2 = ring[i + 1][0] - x0
    const y2 = ring[i + 1][1] - y0
    const cross = x1 * y2 - x2 * y1
    twiceArea += cross
    cx += (x1 + x2) * cross
    cy += (y1 + y2) * cross
    span = Math.max(span, Math.abs(x2), Math.abs(y2))
    minX = Math.min(minX, x2)
    maxX = Math.max(maxX, x2)
    minY = Math.min(minY, y2)
    maxY = Math.max(maxY, y2)
  }
  // Relative to the ring's extent, so a sliver is still a polygon but collinear points are not.
  if (!(Math.abs(twiceArea) > span * span * 1e-12)) return null
  const x = cx / (3 * twiceArea)
  const y = cy / (3 * twiceArea)
  // A simple polygon's centroid is inside its bounding box. One that is not comes from a ring that
  // crosses itself (a "bowtie" outline, a common OSM error that osmium builds no area from): its
  // lobes' signed areas nearly cancel and push the result far away, 663 m for a 20 m outline whose
  // lobes differ by 1%. Null, so centroid() falls back to the middle of the outline.
  const eps = span * 1e-9
  if (!(x >= minX - eps && x <= maxX + eps && y >= minY - eps && y <= maxY + eps)) return null
  return { lon: x0 + x, lat: y0 + y, area: Math.abs(twiceArea) / 2 }
}

/** Length-weighted centre of a line; its first point if it has no length. */
export function lineCentroid(coords) {
  if (!Array.isArray(coords) || !coords.length) return null
  const [x0, y0] = coords[0]
  let total = 0
  let cx = 0
  let cy = 0
  for (let i = 0; i < coords.length - 1; i++) {
    const x1 = coords[i][0] - x0
    const y1 = coords[i][1] - y0
    const x2 = coords[i + 1][0] - x0
    const y2 = coords[i + 1][1] - y0
    const len = Math.hypot(x2 - x1, y2 - y1)
    total += len
    cx += ((x1 + x2) / 2) * len
    cy += ((y1 + y2) / 2) * len
  }
  return total > 0 ? { lon: x0 + cx / total, lat: y0 + cy / total } : { lon: x0, lat: y0 }
}

const isClosed = (c) => c.length >= 4 && c[0][0] === c.at(-1)[0] && c[0][1] === c.at(-1)[1]

/**
 * [lon, lat] of a GeoJSON geometry from `osmium export`: a node's point; the area centroid of a
 * polygon's outer ring (holes ignored), area-weighted over a multipolygon's outer rings; the
 * length-weighted centre of an open way, or of a closed one that crosses itself. osmium writes a
 * closed way twice, as a LineString and as an area: both get the ring's area centroid, so either
 * copy yields the same place.
 */
export function centroid(geometry) {
  const c = geometry?.coordinates
  if (!Array.isArray(c)) return null
  let p = null
  switch (geometry.type) {
    case 'Point':
      p = { lon: c[0], lat: c[1] }
      break
    case 'LineString':
      p = (isClosed(c) && ringCentroid(c)) || lineCentroid(c)
      break
    case 'Polygon':
    case 'MultiPolygon': {
      const outers = (geometry.type === 'Polygon' ? [c] : c).map((polygon) => polygon?.[0]).filter(Array.isArray)
      let area = 0
      let lon = 0
      let lat = 0
      for (const ring of outers) {
        const r = ringCentroid(ring)
        if (!r) continue
        area += r.area
        lon += r.lon * r.area
        lat += r.lat * r.area
      }
      p = area > 0 ? { lon: lon / area, lat: lat / area } : outers.length ? lineCentroid(outers[0]) : null
      break
    }
  }
  return p && Number.isFinite(p.lon) && Number.isFinite(p.lat) ? [p.lon, p.lat] : null
}

// -------------------------------------------------------------------------------------------- places

const TYPES = ['node', 'way', 'relation']
const nonEmpty = (s) => (typeof s === 'string' && s.trim() ? s.trim() : undefined)

/** Mirrors the app's address(from:): "55 Cyril Magnin Street, San Francisco". */
export function formatAddress(tags) {
  const street = [tags['addr:housenumber'], tags['addr:street']].filter((p) => p !== undefined).join(' ')
  const parts = [street, tags['addr:city']].filter((p) => p !== undefined && p !== '')
  return parts.length ? parts.join(', ') : undefined
}

/** An OSM element ({type, id, lat, lon, tags}) as a place, or null if it is unnamed, uncategorised or has no position. */
export function toPlace(element) {
  const tags = element?.tags
  if (!tags || !TYPES.includes(element.type) || !Number.isSafeInteger(element.id)) return null
  const name = nonEmpty(tags.name)
  const category = categoryFor(tags)
  const lat = element.lat ?? element.center?.lat
  const lon = element.lon ?? element.center?.lon
  if (!name || !category || !Number.isFinite(lat) || !Number.isFinite(lon)) return null
  const place = { id: `${element.type}/${element.id}`, name, lat: round6(lat), lon: round6(lon), category }
  const optional = {
    cuisine: nonEmpty(tags.cuisine),
    openingHours: nonEmpty(tags.opening_hours),
    website: nonEmpty(tags.website ?? tags['contact:website']),
    phone: nonEmpty(tags.phone ?? tags['contact:phone']),
    address: formatAddress(tags),
  }
  for (const [k, v] of Object.entries(optional)) if (v !== undefined) place[k] = v
  return place
}

/**
 * A feature of `osmium export -a type,id` as a place: properties are the tags plus "@type" and "@id"
 * (the node, way or relation, also for areas). Name and category are checked before any geometry.
 */
export function featureToPlace(feature) {
  const props = feature?.properties
  if (!props || !nonEmpty(props.name) || !categoryFor(props)) return null
  const { '@type': type, '@id': id, ...tags } = props
  const c = centroid(feature.geometry)
  return c ? toPlace({ type, id, lat: c[1], lon: c[0], tags }) : null
}

/** node < way < relation, then numerically: stable across runs, so weekly diffs stay small. */
export function compareIds(a, b) {
  const [ta, na] = a.id.split('/')
  const [tb, nb] = b.id.split('/')
  return TYPES.indexOf(ta) - TYPES.indexOf(tb) || Number(na) - Number(nb)
}

// --------------------------------------------------------------------------------------- geojsonseq

/** The lines of a byte stream, one at a time: nothing but the current line and chunk is held. */
export async function* lines(stream, { maxLineChars = 64 * 1024 * 1024 } = {}) {
  const decoder = new StringDecoder('utf8')
  let rest = ''
  for await (const chunk of stream) {
    const parts = (rest + (typeof chunk === 'string' ? chunk : decoder.write(chunk))).split('\n')
    rest = parts.pop()
    if (rest.length > maxLineChars) throw new Error(`a line longer than ${maxLineChars} characters`)
    yield* parts
  }
  rest += decoder.end()
  if (rest) yield rest
}

/** GeoJSON Text Sequences (RFC 8142) or plain JSON lines: one feature per non-blank line. */
export async function* readFeatures(stream) {
  let n = 0
  for await (let line of lines(stream)) {
    n++
    if (line.charCodeAt(0) === 0x1e) line = line.slice(1)
    if (!line.trim()) continue
    try {
      yield JSON.parse(line)
    } catch (err) {
      throw new Error(`geojsonseq line ${n}: ${err.message}`)
    }
  }
}

// ----------------------------------------------------------------------------------------- geofabrik

/**
 * fatal: Geofabrik refused us (403/429) and the run must stop asking. retryable: worth one more try.
 * transfer: the bytes did not arrive (network, 5xx, stall, too slow, md5); refresh() stops the run
 * after TRANSFER_BREAKER such extracts in a row.
 */
export class GeofabrikError extends Error {
  constructor(message, { status, retryable = false, fatal = false, transfer = retryable } = {}) {
    super(message)
    this.name = 'GeofabrikError'
    this.status = status
    this.retryable = retryable
    this.fatal = fatal
    this.transfer = transfer
  }
}

const errorText = (err) => String(err?.message ?? err).slice(0, 300)

const kBps = (bytesPerSec) => Math.round(bytesPerSec / 1000)

/** Where a download stands: the bytes in the file, their md5 so far, the whole length, and the validator for If-Range. */
const freshPart = () => ({ bytes: 0, hash: createHash('md5'), length: null, validator: null })
const resetPart = (part) => Object.assign(part, freshPart())

/**
 * Downloads extracts, one at a time (the caller awaits each). <path>-latest.osm.pbf is a redirect to
 * a dated file, e.g. district-of-columbia-260925.osm.pbf, whose .md5 never changes: the download is
 * checked against the md5 of that same file, never of a newer "latest", and a broken transfer can
 * resume from the same file with a Range request.
 */
export function createGeofabrik({
  fetch = globalThis.fetch,
  baseUrl = GEOFABRIK_URL,
  log = () => {},
  signal = new AbortController().signal,
  sleep = (ms) => delay(ms, undefined, { signal }),
  now = () => Date.now(),
  deadlineAt = Infinity,
  cutoffAt = Infinity,
  retryWaitMs = RETRY_WAIT_MS,
  stallMs = STALL_MS,
  slowWindowMs = SLOW_WINDOW_MS,
  minBytesPerSec = MIN_BYTES_PER_SEC,
  maxBytes = MAX_EXTRACT_BYTES,
} = {}) {
  // downloadBytes: every byte received, also those of a transfer that failed; the traffic we cost.
  const stats = { requests: 0, downloads: 0, downloadBytes: 0, retries: 0, resumed: 0 }
  const origin = new URL(baseUrl).origin

  async function request(url, init = {}) {
    stats.requests++
    try {
      return await fetch(url, { ...init, headers: { 'User-Agent': USER_AGENT, ...init.headers }, signal: init.signal ?? AbortSignal.any([signal, AbortSignal.timeout(60_000)]) })
    } catch (err) {
      if (signal.aborted) throw err
      throw new GeofabrikError(`${url}: ${errorText(err)}`, { retryable: true })
    }
  }

  const httpError = (url, status) =>
    new GeofabrikError(`${url}: HTTP ${status}`, { status, retryable: status >= 500, fatal: status === 403 || status === 429 })

  /** The dated file that <path>-latest.osm.pbf redirects to: one HEAD, no body from Geofabrik. */
  async function resolve(path) {
    const latest = `${baseUrl}/${path}-latest.osm.pbf`
    const res = await request(latest, { method: 'HEAD', redirect: 'manual' })
    await res.body?.cancel()
    if (res.status === 200) return latest
    const location = res.headers.get('location')
    if (res.status < 300 || res.status >= 400 || !location) throw httpError(latest, res.status)
    const target = new URL(location, latest)
    const dir = latest.slice(0, latest.lastIndexOf('/') + 1)
    const name = path.slice(path.lastIndexOf('/') + 1)
    const file = target.href.slice(dir.length)
    if (target.origin !== origin || !target.href.startsWith(dir) || !file.startsWith(`${name}-`) || !/^[\w.-]+\.osm\.pbf$/.test(file)) {
      throw new GeofabrikError(`${latest} redirects to an unexpected place: ${target.href}`)
    }
    return target.href
  }

  async function md5Of(url) {
    const res = await request(`${url}.md5`)
    const text = await res.text()
    if (!res.ok) throw httpError(`${url}.md5`, res.status)
    const m = /^([0-9a-f]{32})\s/i.exec(`${text.trim()} `)
    if (!m) throw new GeofabrikError(`${url}.md5: no md5 in ${JSON.stringify(text.slice(0, 80))}`, { retryable: true })
    return m[1].toLowerCase()
  }

  /**
   * Streams `url` into `dest`, carrying on from `part`. With bytes already there it asks for the
   * rest (Range, If-Range with the first response's validator) and accepts only a 206 that starts
   * exactly there; a 200 means the server sent the whole file again, which then starts over.
   */
  async function fetchTo(url, dest, part) {
    const stop = new AbortController()
    const timer = setTimeout(() => stop.abort(new GeofabrikError(`${url}: no data for ${stallMs / 1000} s`, { retryable: true })), stallMs)
    try {
      const resume = part.bytes > 0
      const headers = resume ? { Range: `bytes=${part.bytes}-`, ...(part.validator ? { 'If-Range': part.validator } : {}) } : {}
      const res = await request(url, { headers, signal: AbortSignal.any([signal, stop.signal]) })
      if (resume && res.status === 206) {
        const range = /^bytes (\d+)-\d+\/(\d+)$/.exec(res.headers.get('content-range') ?? '')
        if (!range || Number(range[1]) !== part.bytes || (part.length !== null && Number(range[2]) !== part.length)) {
          await res.body?.cancel()
          const asked = part.bytes
          resetPart(part)
          throw new GeofabrikError(`${url}: asked for bytes ${asked}- and got Content-Range ${res.headers.get('content-range')}`, { retryable: true })
        }
        part.length = Number(range[2])
        stats.resumed++
      } else if (res.status !== 200) {
        await res.body?.cancel()
        throw httpError(url, res.status)
      } else {
        if (resume) log('warn', 'download_restart', { url, had: part.bytes })
        resetPart(part)
        const length = Number(res.headers.get('content-length'))
        const etag = res.headers.get('etag')
        part.length = length > 0 ? length : null
        part.validator = etag && !etag.startsWith('W/') ? etag : res.headers.get('last-modified')
      }
      if (part.length > maxBytes) {
        await res.body?.cancel()
        throw new GeofabrikError(`${url}: ${part.length} bytes, more than the ${maxBytes} the work volume is sized for`)
      }
      const fh = await open(dest, part.bytes > 0 ? 'r+' : 'w')
      try {
        // Whatever a failed write left past the hashed bytes goes.
        if (part.bytes > 0) await fh.truncate(part.bytes)
        let windowAt = now()
        let windowBytes = part.bytes
        for await (const chunk of res.body) {
          timer.refresh()
          stats.downloadBytes += chunk.length
          if (part.bytes + chunk.length > maxBytes) throw new GeofabrikError(`${url}: more than ${maxBytes} bytes`)
          await fh.write(chunk, 0, chunk.length, part.bytes)
          part.hash.update(chunk)
          part.bytes += chunk.length
          const t = now()
          if (t > cutoffAt) throw new GeofabrikError(`${url}: still downloading at the transfer cutoff (${Math.round(part.bytes / 1e6)} MB in)`, { transfer: true })
          if (t - windowAt < slowWindowMs) continue
          const perSec = ((part.bytes - windowBytes) * 1000) / (t - windowAt)
          if (perSec < minBytesPerSec) {
            throw new GeofabrikError(`${url}: ${kBps(perSec)} kB/s over ${Math.round((t - windowAt) / 1000)} s, below ${kBps(minBytesPerSec)} kB/s`, { retryable: true })
          }
          if (part.length !== null && t + ((part.length - part.bytes) * 1000) / perSec > cutoffAt) {
            throw new GeofabrikError(`${url}: at ${kBps(perSec)} kB/s the other ${Math.round((part.length - part.bytes) / 1e6)} MB would not arrive in time`, { transfer: true })
          }
          windowAt = t
          windowBytes = part.bytes
        }
      } finally {
        await fh.close()
      }
      if (part.length !== null && part.bytes !== part.length) throw new GeofabrikError(`${url}: got ${part.bytes} of ${part.length} bytes`, { retryable: true })
    } catch (err) {
      if (signal.aborted) throw err
      if (stop.signal.aborted) throw stop.signal.reason
      if (err instanceof GeofabrikError) throw err
      throw new GeofabrikError(`${url}: ${errorText(err)}`, { retryable: true })
    } finally {
      clearTimeout(timer)
    }
  }

  /**
   * The extract at `path` into `dest`, md5-checked. Tried twice at most: the second try goes to the
   * same dated file and resumes from the bytes the first one got, unless their md5 was wrong.
   */
  async function download(path, dest) {
    const part = freshPart()
    let url = null
    let expected = null
    let ms = 0
    for (let attempt = 1; ; attempt++) {
      const started = now()
      let error = null
      try {
        url ??= await resolve(path)
        expected ??= await md5Of(url)
        await fetchTo(url, dest, part)
      } catch (err) {
        error = err
      }
      ms += now() - started
      if (!error) {
        const md5 = part.hash.digest('hex')
        if (md5 === expected) {
          stats.downloads++
          return { url, file: url.slice(url.lastIndexOf('/') + 1), bytes: part.bytes, md5, ms }
        }
        resetPart(part)
        error = new GeofabrikError(`${url}: md5 ${md5}, expected ${expected}`, { retryable: true })
      }
      if (signal.aborted || !error?.retryable || attempt >= 2 || now() + retryWaitMs > deadlineAt) {
        await rm(dest, { force: true })
        throw error
      }
      stats.retries++
      log('warn', 'download_retry', { path, attempt, waitMs: retryWaitMs, resumeFrom: part.bytes, error: errorText(error) })
      await sleep(retryWaitMs)
    }
  }

  return { resolve, download, stats }
}

// -------------------------------------------------------------------------------------------- osmium

// OPL (osmium's line format, written with -f opl,add_metadata=false) looks like
//   n281074 T x-77.1082382 y38.9366569
//   w100 Tamenity=cafe,name=Square%20%Cafe Nn2,n3,n4,n5,n2
//   r200 Ttype=multipolygon,amenity=bar Mw102@outer,w103@inner
// Fields are separated by spaces, and spaces, commas and '=' inside tags are escaped, so a field
// never contains a raw space. Lines are handled as latin1 (one char per byte) and written back the
// same way: tag bytes pass through untouched, and only ASCII ids are ever read here.
const NODE = 0x6e // 'n'
const WAY = 0x77 // 'w'
const RELATION = 0x72 // 'r'

function digitsAt(s, i) {
  let id = 0
  for (let c = s.charCodeAt(i); c >= 48 && c <= 57; c = s.charCodeAt(++i)) id = id * 10 + (c - 48)
  return id
}

/** The OPL field starting with `marker` (' T', ' N', ' M'), without the marker; '' if absent. */
function oplField(line, marker) {
  const at = line.indexOf(marker)
  if (at === -1) return ''
  const end = line.indexOf(' ', at + 2)
  return line.slice(at + 2, end === -1 ? line.length : end)
}

/** Pushes the node ids a way line refers to (Nn1,n2,...) onto `into`. */
export function wayNodes(line, into = []) {
  const refs = oplField(line, ' N')
  for (let i = 0; i < refs.length; i = refs.indexOf(',', i) + 1 || refs.length) {
    if (refs.charCodeAt(i) === NODE) into.push(digitsAt(refs, i + 1))
  }
  return into
}

/** Pushes the way members of a multipolygon or boundary relation line (Mw1@outer,...) onto `into`. */
export function areaMembers(line, into = []) {
  const tags = oplField(line, ' T').split(',')
  if (!tags.includes('type=multipolygon') && !tags.includes('type=boundary')) return into
  const members = oplField(line, ' M')
  for (let i = 0; i < members.length; i = members.indexOf(',', i) + 1 || members.length) {
    if (members.charCodeAt(i) === WAY) into.push(digitsAt(members, i + 1))
  }
  return into
}

/** Sorted, without duplicates: 8 bytes an id, where a Set of ids above 2^31 takes about five times that. */
export function idList(ids) {
  const sorted = Float64Array.from(ids).sort()
  let n = 0
  for (let i = 0; i < sorted.length; i++) if (i === 0 || sorted[i] !== sorted[n - 1]) sorted[n++] = sorted[i]
  return sorted.subarray(0, n)
}

/**
 * keep(type, id) for scanOpl over one object type in ascending id order: a cursor over a sorted id
 * list, so a lookup is a comparison, not a hash. Geofabrik's extracts are sorted by type, then id; a
 * stream that is not would silently lose objects, so it throws instead.
 */
export function sortedMatcher(ids) {
  let at = 0
  let last = -Infinity
  return (type, id) => {
    if (id < last) throw new Error(`OPL ids out of order (${id} after ${last}): the extract is not sorted`)
    last = id
    while (at < ids.length && ids[at] < id) at++
    return at < ids.length && ids[at] === id
  }
}

/**
 * Calls onLine(line, type, id) for each OPL line of a byte stream whose type and id `keep` accepts.
 * type is the byte 'n', 'w' or 'r'. This loop sees every node of an extract, over 100 million for the
 * big ones, so it works on the raw bytes: a line that is not kept is never copied or decoded.
 */
export async function scanOpl(stream, keep, onLine, { maxLineBytes = 64 * 1024 * 1024 } = {}) {
  const each = (buf, start, end) => {
    if (end <= start) return
    let id = 0
    for (let i = start + 1, c = buf[i]; c >= 48 && c <= 57; c = buf[++i]) id = id * 10 + (c - 48)
    const type = buf[start]
    if (keep(type, id)) onLine(buf.toString('latin1', start, end), type, id)
  }
  let rest = null
  for await (let chunk of stream) {
    if (typeof chunk === 'string') chunk = Buffer.from(chunk, 'latin1')
    let start = 0
    if (rest) {
      const nl = chunk.indexOf(10)
      if (nl === -1) {
        rest = Buffer.concat([rest, chunk])
        if (rest.length > maxLineBytes) throw new Error(`an OPL line longer than ${maxLineBytes} bytes`)
        continue
      }
      const line = Buffer.concat([rest, chunk.subarray(0, nl)])
      each(line, 0, line.length)
      rest = null
      start = nl + 1
    }
    for (let nl = chunk.indexOf(10, start); nl !== -1; nl = chunk.indexOf(10, start)) {
      each(chunk, start, nl)
      start = nl + 1
    }
    if (start < chunk.length) rest = Buffer.from(chunk.subarray(start))
  }
  if (rest) each(rest, 0, rest.length)
}

/** Appends lines to a file with plain synchronous writes, batched; the bytes are the latin1 chars. */
async function oplWriter(path) {
  const fh = await open(path, 'w')
  let buf = []
  let size = 0
  const flush = () => {
    if (!size) return
    writeSync(fh.fd, buf.join(''), null, 'latin1')
    buf = []
    size = 0
  }
  return {
    line(line) {
      buf.push(line, '\n')
      size += line.length + 1
      if (size >= 1 << 20) flush()
    },
    async close() {
      try {
        flush()
      } finally {
        await fh.close()
      }
    },
  }
}

/** osmium-tool as a child process. Every call rejects unless osmium exits 0; SIGTERM reaches it via `signal`. */
export function createOsmium({ bin = 'osmium', signal, env = process.env } = {}) {
  async function run(args, consume) {
    const child = spawn(bin, args, { stdio: ['ignore', consume ? 'pipe' : 'ignore', 'pipe'], env, signal })
    let stderr = ''
    child.stderr.setEncoding('utf8')
    child.stderr.on('data', (d) => (stderr = (stderr + d).slice(-4000)))
    const exited = new Promise((resolve) => {
      child.once('error', (error) => resolve({ error }))
      child.once('close', (code, sig) => resolve({ code, sig }))
    })
    let result
    try {
      if (consume) result = await consume(child.stdout)
    } catch (err) {
      child.kill('SIGTERM')
      await exited
      throw err
    }
    const { error, code, sig } = await exited
    if (error) throw error
    if (code !== 0) {
      const tail = stderr.trim().split('\n').slice(-3).join(' | ').slice(0, 400)
      throw new Error(`osmium ${args[0]} ${code === null ? `killed by ${sig}` : `exited ${code}`}${tail ? `: ${tail}` : ''}`)
    }
    return result
  }
  const text = (args) =>
    run(args, async (stdout) => {
      let out = ''
      for await (const chunk of stdout.setEncoding('utf8')) out += chunk
      return out
    })
  const opl = (args, keep, onLine) => run([...args, '-f', 'opl,add_metadata=false', '--no-progress'], (stdout) => scanOpl(stdout, keep, onLine))
  const all = () => true

  return {
    async version() {
      return (await text(['--version'])).split('\n')[0].trim()
    },
    /** The extract's replication timestamp from its header: how current the OSM data in it is. */
    async timestamp(file) {
      const option = JSON.parse(await text(['fileinfo', '-j', file]))?.header?.option ?? {}
      return option.osmosis_replication_timestamp ?? option.timestamp ?? null
    },
    /**
     * Every object in `pbf` with a mapped tag, as a GeoJSON feature from `osmium export -a type,id`,
     * to onFeature. `osmium tags-filter` without -R would add the nodes and members these objects
     * need itself, but it tracks their ids in bitmaps over the whole id range, about 1.7 GB for any
     * big extract (NOTES.md). The same objects, in a few MB of sorted id lists:
     *   1. `tags-filter -R`: the matching nodes, ways and relations, without what they refer to;
     *   2. only if multipolygons matched: `cat -t way`, keeping their member ways;
     *   3. `cat -t node`, keeping the matching nodes and every node those ways need;
     * all in order into a small OPL file, which `osmium export` turns into points, lines and areas.
     * Export writes relations only as areas, and only multipolygons and boundaries it can assemble;
     * a way needs two located nodes. What matched but came out as nothing is counted in
     * `withoutGeometry` (a type=site restaurant relation, a multipolygon with a broken ring).
     */
    async features(pbf, workDir, onFeature) {
      const t0 = Date.now()
      let nodes = []
      const ways = new Map()
      const relations = []
      const relationIds = []
      const memberIds = []
      await opl(['tags-filter', pbf, ...tagFilters(), '--omit-referenced'], all, (line, type, id) => {
        if (type === NODE) nodes.push(id)
        else if (type === WAY) ways.set(id, line)
        else if (type === RELATION) {
          relations.push(line)
          relationIds.push(id)
          areaMembers(line, memberIds)
        }
      })
      const matched = { nodes: nodes.length, ways: ways.size, relations: relations.length }
      const matchedWays = [...ways.keys()]
      const members = idList(memberIds.filter((id) => !ways.has(id)))
      const t1 = Date.now()
      if (members.length) await opl(['cat', pbf, '-t', 'way'], sortedMatcher(members), (line, type, id) => ways.set(id, line))
      for (const line of ways.values()) wayNodes(line, nodes)
      nodes = idList(nodes)
      const t2 = Date.now()
      const small = join(workDir, 'objects.opl')
      const out = await oplWriter(small)
      try {
        await opl(['cat', pbf, '-t', 'node'], sortedMatcher(nodes), (line) => out.line(line))
        for (const id of [...ways.keys()].sort((a, b) => a - b)) out.line(ways.get(id))
        for (const line of relations) out.line(line)
      } finally {
        await out.close()
      }
      const t3 = Date.now()
      const objectsBytes = (await stat(small)).size
      let features = 0
      const exportedWays = new Set()
      const exportedRelations = new Set()
      await run(['export', small, '-f', 'geojsonseq', '-x', 'print_record_separator=false', '-a', 'type,id', '--no-progress'], async (stdout) => {
        for await (const feature of readFeatures(stdout)) {
          features++
          const type = feature?.properties?.['@type']
          if (type === 'way') exportedWays.add(feature.properties['@id'])
          else if (type === 'relation') exportedRelations.add(feature.properties['@id'])
          onFeature(feature)
        }
      })
      await rm(small, { force: true })
      const withoutGeometry = {
        ways: matchedWays.filter((id) => !exportedWays.has(id)).length,
        relations: relationIds.filter((id) => !exportedRelations.has(id)).length,
      }
      return { matched, withoutGeometry, memberWays: members.length, nodes: nodes.length, objectsBytes, features, filterMs: t1 - t0, waysMs: t2 - t1, nodesMs: t3 - t2, exportMs: Date.now() - t3 }
    },
  }
}

// ------------------------------------------------------------------------------------------- storage

export const sha256hex = (buf) => createHash('sha256').update(buf).digest('hex')

/** One place per line: valid JSON that diffs by place. */
export function serializeCity({ cityId, generatedAt, osmBase, places }) {
  const head = JSON.stringify({ cityId, generatedAt, ...LICENSE, ...(osmBase ? { osmBase } : {}) }).slice(0, -1)
  const body = places.length ? `\n${places.map((p) => JSON.stringify(p)).join(',\n')}\n` : ''
  return `${head},"places":[${body}]}\n`
}

/** Temp file in the same directory, fsync, rename: readers see the old file or the new one, never half.
 *  The temp name is random, not the pid: in a container every run is pid 1. */
export async function writeAtomic(path, data) {
  const tmp = `${path}.tmp-${randomUUID()}`
  try {
    const fh = await open(tmp, 'w', 0o644)
    try {
      await fh.writeFile(data)
      await fh.sync()
    } finally {
      await fh.close()
    }
    await rename(tmp, path)
  } catch (err) {
    await unlink(tmp).catch(() => {})
    throw err
  }
}

/** The .gz first, then the .json: the server only serves a .gz whose content matches the .json. */
export async function writeCity(dataDir, city) {
  const json = Buffer.from(serializeCity(city))
  const gz = gzipSync(json, { level: 9 })
  await writeAtomic(join(dataDir, `${city.cityId}.json.gz`), gz)
  await writeAtomic(join(dataDir, `${city.cityId}.json`), json)
  return { sha256: sha256hex(json), bytes: json.length, gzipBytes: gz.length }
}

export function serializeIndex(entries, cityOrder) {
  const cities = cityOrder
    .filter((id) => entries.has(id))
    .map((id) => {
      const e = entries.get(id)
      return { cityId: id, generatedAt: e.generatedAt, count: e.count, bytes: e.bytes, gzipBytes: e.gzipBytes, sha256: e.sha256, stationsSha256: e.stationsSha256 }
    })
  return `${JSON.stringify({ ...LICENSE, cities }, null, 1)}\n`
}

/** What is on disk now, verified by hashing: index.json only contributes stationsSha256. */
export async function loadEntries(dataDir, cityIds, log = () => {}) {
  let previous = new Map()
  try {
    const idx = JSON.parse(await readFile(join(dataDir, 'index.json'), 'utf8'))
    previous = new Map(idx.cities.map((c) => [c.cityId, c]))
  } catch (err) {
    if (err.code !== 'ENOENT') log('warn', 'index_unreadable', { error: String(err.message).slice(0, 200) })
  }
  const entries = new Map()
  for (const cityId of cityIds) {
    let json
    try {
      json = await readFile(join(dataDir, `${cityId}.json`))
    } catch (err) {
      if (err.code === 'ENOENT') continue
      throw err
    }
    try {
      const parsed = JSON.parse(json)
      const sha256 = sha256hex(json)
      const gzipBytes = (await stat(join(dataDir, `${cityId}.json.gz`)).catch(() => null))?.size ?? null
      const prev = previous.get(cityId)
      entries.set(cityId, {
        generatedAt: parsed.generatedAt,
        count: parsed.places.length,
        bytes: json.length,
        gzipBytes,
        sha256,
        stationsSha256: prev?.sha256 === sha256 ? prev.stationsSha256 : null,
      })
    } catch (err) {
      log('warn', 'city_file_unreadable', { cityId, error: String(err.message).slice(0, 200) })
    }
  }
  return entries
}

/**
 * One refresh at a time. The CronJob's concurrencyPolicy does not cover a manual
 * `kubectl create job --from=cronjob/...`, and two runs would download every extract twice. The
 * holder touches the lock every minute; a lock untouched for staleMs belongs to a killed pod
 * (OOM, SIGKILL) and is taken over. A second run waits up to waitMs, then gives up.
 */
export async function acquireLock(dataDir, { staleMs = 5 * 60_000, waitMs = 6 * 60_000, pollMs = 15_000, log = () => {} } = {}) {
  const path = join(dataDir, 'refresh.lock')
  const me = JSON.stringify({ host: hostname(), pid: process.pid, id: randomUUID(), since: new Date().toISOString() })
  const giveUpAt = Date.now() + waitMs
  for (;;) {
    try {
      const fh = await open(path, 'wx', 0o644)
      await fh.writeFile(me)
      await fh.close()
      break
    } catch (err) {
      if (err.code !== 'EEXIST') throw err
    }
    const st = await stat(path).catch(() => null)
    if (!st) continue
    const holder = await readFile(path, 'utf8').catch(() => '?')
    if (Date.now() - st.mtimeMs > staleMs) {
      log('warn', 'lock_stale', { holder, idleS: Math.round((Date.now() - st.mtimeMs) / 1000) })
      await unlink(path).catch(() => {})
      continue
    }
    if (Date.now() >= giveUpAt) throw new Error(`another refresh is running: ${holder}`)
    log('info', 'lock_wait', { holder })
    await delay(pollMs)
  }
  const heartbeat = setInterval(() => utimes(path, new Date(), new Date()).catch(() => {}), 60_000)
  heartbeat.unref()
  return {
    async release() {
      clearInterval(heartbeat)
      if ((await readFile(path, 'utf8').catch(() => null)) === me) await unlink(path).catch(() => {})
    },
  }
}

export function loadStations(text) {
  const cities = JSON.parse(text)?.cities
  if (!cities || typeof cities !== 'object' || !Object.keys(cities).length) throw new Error('stations: no cities')
  for (const [id, list] of Object.entries(cities)) {
    if (!/^[a-z0-9][a-z0-9-]{0,31}$/.test(id) || id === 'index') throw new Error(`stations: bad city id ${id}`)
    if (!Array.isArray(list) || !list.length) throw new Error(`stations: ${id} has no stations`)
    for (const p of list) {
      if (!Array.isArray(p) || p.length !== 2 || !p.every(Number.isFinite) || Math.abs(p[0]) > 90 || Math.abs(p[1]) > 180) {
        throw new Error(`stations: ${id} has a bad coordinate ${JSON.stringify(p)}`)
      }
    }
  }
  return cities
}

// ---------------------------------------------------------------------------------------------- run

const isoSeconds = (ms) => new Date(ms).toISOString().replace(/\.\d{3}Z$/, 'Z')

/**
 * Peak memory of the container (cgroup v2, osmium included). `anon` is what the limit is really
 * about, and the kernel keeps no peak of it, so it is sampled every 250 ms (a shorter spike can be
 * missed). memory.peak also counts the downloads' page cache, which the kernel reclaims under the
 * limit. Both are null outside a container.
 */
export function watchMemory(intervalMs = 250) {
  let anonPeak = null
  const sample = () =>
    readFile('/sys/fs/cgroup/memory.stat', 'utf8').then(
      (text) => {
        const m = /^anon (\d+)$/m.exec(text)
        if (m) anonPeak = Math.max(anonPeak ?? 0, Number(m[1]))
      },
      () => {},
    )
  const timer = setInterval(sample, intervalMs)
  timer.unref()
  sample()
  const mb = (bytes) => (bytes === null ? null : Math.round(bytes / 2 ** 20))
  return {
    async snapshot() {
      await sample()
      const peak = await readFile('/sys/fs/cgroup/memory.peak', 'utf8').catch(() => '')
      return { anonPeakMB: mb(anonPeak), cgroupPeakMB: mb(/^\d+\s*$/.test(peak) ? Number(peak) : null), nodeMaxRssMB: mb(process.resourceUsage().maxRSS * 1024) }
    },
    stop: () => clearInterval(timer),
  }
}

/**
 * One run. Every -latest link is checked first (HEAD). Then the cities are taken stalest first
 * (never built, then oldest generatedAt, then stations.json order), so a run that cannot do them all
 * leaves a different tail each week; a city is started only if MIN_EXTRACT_MS per extract it still
 * needs fits before the deadline, else it waits for the next run. Each extract is processed once,
 * for every city that needs it, and a city is written as soon as the last of its extracts is in.
 * `stop` says why the run ended early, or left cities out: breaker (Geofabrik refused us),
 * transfers (TRANSFER_BREAKER extracts in a row failed to arrive), deadline or aborted (SIGTERM);
 * done otherwise.
 */
export async function refresh({
  dataDir,
  workDir,
  stations,
  only = [],
  extracts = EXTRACTS,
  geofabrik,
  osmium,
  log = () => {},
  now = () => Date.now(),
  allowShrink = false,
  deadlineAt = Infinity,
  signal = new AbortController().signal,
  minExtractMs = MIN_EXTRACT_MS,
  lock = {},
}) {
  const started = now()
  // `dc dc` is one city: a repeated id would finish the same city twice.
  only = [...new Set(only)]
  for (const id of only) if (!stations[id]) throw new Error(`unknown city ${id}`)
  await mkdir(dataDir, { recursive: true })
  await mkdir(workDir, { recursive: true })
  const held = await acquireLock(dataDir, { log, ...lock })
  const runDir = join(workDir, `refresh-${randomUUID()}`)
  const memory = watchMemory()
  try {
    // Leftovers of a run that died: temp files of the atomic writes, and work dirs. Safe: we hold the lock.
    for (const f of await readdir(dataDir)) if (/\.tmp-[\w-]+$/.test(f)) await unlink(join(dataDir, f)).catch(() => {})
    for (const f of await readdir(workDir)) if (/^refresh-[\w-]+$/.test(f)) await rm(join(workDir, f), { recursive: true, force: true })
    // The Overpass tile cache of the job this one replaced.
    await rm(join(dataDir, 'tiles'), { recursive: true, force: true })
    await mkdir(runDir)
    const cityOrder = Object.keys(stations)
    const targets = only.length ? only : cityOrder
    return await refreshLocked({ dataDir, runDir, stations, cityOrder, targets, extracts, geofabrik, osmium, log, now, allowShrink, deadlineAt, signal, minExtractMs, memory, started })
  } finally {
    memory.stop()
    await rm(runDir, { recursive: true, force: true })
    await held.release()
  }
}

async function refreshLocked(o) {
  const { dataDir, runDir, stations, cityOrder, targets, extracts, geofabrik, osmium, log, now, allowShrink, deadlineAt, signal, minExtractMs, memory, started } = o
  const entries = await loadEntries(dataDir, cityOrder, log)
  const writeIndex = () => writeAtomic(join(dataDir, 'index.json'), serializeIndex(entries, cityOrder))
  await writeIndex()

  const stationsSha = Object.fromEntries(cityOrder.map((id) => [id, sha256hex(JSON.stringify(stations[id])).slice(0, 16)]))
  // Per target city: the extracts still to come, the places so far, and how it ended.
  const cities = new Map(
    targets.map((cityId) => {
      const paths = extracts[cityId] ?? []
      return [cityId, { cityId, paths, pending: new Set(paths), index: null, places: new Map(), osmBase: null, status: 'pending', error: paths.length ? null : 'no Geofabrik extracts configured for this city (EXTRACTS)' }]
    }),
  )
  const order = [...new Set(targets.flatMap((id) => extracts[id] ?? []))]
  const summary = {
    stop: 'done',
    breaker: null,
    extracts: { total: order.length, ok: [], failed: [], skipped: [] },
    cities: { rebuilt: [], unchanged: [], failed: [] },
  }

  async function processExtract(path, users) {
    const pbf = join(runDir, `${path.slice(path.lastIndexOf('/') + 1)}.osm.pbf`)
    try {
      const got = await geofabrik.download(path, pbf)
      const osmBase = await osmium.timestamp(pbf)
      const kept = Object.fromEntries(users.map((c) => [c.cityId, 0]))
      const steps = await osmium.features(pbf, runDir, (feature) => {
        const place = featureToPlace(feature)
        if (!place) return
        for (const c of users) {
          // First copy wins: a closed way comes as a line and as an area (same centroid), and an
          // object on a border can be in two extracts.
          if (c.places.has(place.id) || !c.index.within(place.lat, place.lon)) continue
          c.places.set(place.id, place)
          kept[c.cityId]++
        }
      })
      return { file: got.file, bytes: got.bytes, osmBase, downloadMs: got.ms, ...steps, kept }
    } finally {
      // The download is the disk peak; it is gone before the next one starts.
      await rm(pbf, { force: true })
    }
  }

  async function finishCity(c) {
    if (c.status !== 'pending') return
    const entry = entries.get(c.cityId)
    try {
      const places = [...c.places.values()].sort(compareIds)
      c.places = null
      if (!places.length) throw new Error('no places at all: refusing to publish an empty city')
      if (entry && !allowShrink && entry.count >= 100 && places.length < entry.count / 2) {
        throw new Error(`place count would fall from ${entry.count} to ${places.length}; set ALLOW_SHRINK=1 if that is real`)
      }
      // The same extracts again (a second run on the same day): keep the file, and its sha256, so
      // no app downloads the city again for nothing.
      if (entry && sha256hex(serializeCity({ cityId: c.cityId, generatedAt: entry.generatedAt, osmBase: c.osmBase, places })) === entry.sha256) {
        c.status = 'unchanged'
        if (entry.stationsSha256 !== stationsSha[c.cityId]) {
          entries.set(c.cityId, { ...entry, stationsSha256: stationsSha[c.cityId] })
          await writeIndex()
        }
        summary.cities.unchanged.push(c.cityId)
        log('info', 'city_unchanged', { cityId: c.cityId, places: places.length, osmBase: c.osmBase })
        return
      }
      const generatedAt = isoSeconds(now())
      const written = await writeCity(dataDir, { cityId: c.cityId, generatedAt, osmBase: c.osmBase, places })
      entries.set(c.cityId, { generatedAt, count: places.length, ...written, stationsSha256: stationsSha[c.cityId] })
      await writeIndex()
      c.status = 'rebuilt'
      summary.cities.rebuilt.push(c.cityId)
      log('info', 'city_ok', { cityId: c.cityId, places: places.length, previous: entry?.count ?? null, bytes: written.bytes, gzipBytes: written.gzipBytes, osmBase: c.osmBase, extracts: c.paths })
    } catch (err) {
      c.status = 'failed'
      c.error = errorText(err)
    }
  }

  // The cities still waiting for `path`: not failed, not finished.
  const waiting = (path) => targets.map((id) => cities.get(id)).filter((c) => c.status === 'pending' && !c.error && c.pending.has(path))
  const failExtract = (path, users, err, fields = {}) => {
    const error = errorText(err)
    summary.extracts.failed.push(path)
    for (const c of users) {
      c.error = `extract ${path}: ${error}`
      c.places = null
    }
    log('error', 'extract_fail', { path, error, cities: users.map((c) => c.cityId), ...fields })
    if (err?.fatal) {
      summary.stop = 'breaker'
      summary.breaker = error
      log('error', 'geofabrik_stop', { path, error, status: err.status ?? null })
    }
    return error
  }

  // 1. Every -latest link first, HEAD only (at most one per extract). A region Geofabrik renamed or
  // retired (404) fails its cities before any of their other extracts is downloaded, and a refusal
  // stops the run before any GET. A 5xx or network error is left to the download, which asks again.
  for (const path of order) {
    const users = waiting(path)
    if (!users.length) continue
    if (signal.aborted) {
      summary.stop = 'aborted'
      break
    }
    if (now() + minExtractMs > deadlineAt) break
    try {
      await geofabrik.resolve(path)
    } catch (err) {
      if (signal.aborted) {
        summary.stop = 'aborted'
        break
      }
      if (err?.retryable) {
        log('warn', 'preflight_unresolved', { path, error: errorText(err) })
        continue
      }
      failExtract(path, users, err, { preflight: true })
      if (summary.stop === 'breaker') break
    }
  }

  // 2. The cities, stalest first ('' = never built sorts first; the sort is stable).
  const age = (id) => entries.get(id)?.generatedAt ?? ''
  const queue = summary.stop === 'done' ? [...targets].sort((a, b) => (age(a) < age(b) ? -1 : age(a) > age(b) ? 1 : 0)) : []
  let transferFailures = 0
  cityLoop: for (const cityId of queue) {
    const city = cities.get(cityId)
    if (city.status !== 'pending' || city.error) continue
    if (signal.aborted) {
      summary.stop = 'aborted'
      break
    }
    // All of a city's extracts or none: extracts downloaded for a city the deadline then cuts off
    // would be thrown away. A smaller city further down may still fit.
    const remaining = city.paths.filter((p) => city.pending.has(p))
    const leftMin = Math.max(0, Math.floor((deadlineAt - now()) / 60_000))
    if (now() + remaining.length * minExtractMs > deadlineAt) {
      city.error = `left for the next run: ${remaining.length} extract(s) to go and ${leftMin} min before the deadline`
      city.places = null
      summary.stop = 'deadline'
      log('warn', 'city_deferred', { cityId, extracts: remaining, leftMin })
      continue
    }
    for (const path of remaining) {
      const users = waiting(path)
      // This city lost an earlier extract; any other city that needs this one comes up in its turn.
      if (!users.includes(city)) break
      if (signal.aborted) {
        summary.stop = 'aborted'
        break cityLoop
      }
      if (now() + minExtractMs > deadlineAt) {
        summary.stop = 'deadline'
        break cityLoop
      }
      for (const c of users) c.index ??= stationIndex(stations[c.cityId])
      const t = now()
      try {
        const r = await processExtract(path, users)
        for (const c of users) {
          c.pending.delete(path)
          if (typeof r.osmBase === 'string' && (c.osmBase === null || r.osmBase < c.osmBase)) c.osmBase = r.osmBase
        }
        summary.extracts.ok.push(path)
        transferFailures = 0
        const downloadMBps = r.downloadMs > 0 ? Math.round(r.bytes / r.downloadMs / 100) / 10 : null
        log('info', 'extract_ok', { path, ...r, downloadMBps, ms: now() - t, ...(await memory.snapshot()) })
      } catch (err) {
        if (signal.aborted) {
          summary.stop = 'aborted'
          break cityLoop
        }
        const error = failExtract(path, users, err, { ms: now() - t })
        if (summary.stop === 'breaker') break cityLoop
        transferFailures = err?.transfer ? transferFailures + 1 : 0
        if (transferFailures >= TRANSFER_BREAKER) {
          summary.stop = 'transfers'
          summary.breaker = error
          log('error', 'geofabrik_stop', { path, error, reason: `${transferFailures} extracts in a row did not arrive` })
          break cityLoop
        }
      }
      for (const c of users) if (!c.error && !c.pending.size) await finishCity(c)
    }
  }

  for (const id of targets) {
    const c = cities.get(id)
    if (c.status === 'rebuilt' || c.status === 'unchanged') continue
    const error = c.error ?? `not reached: the run stopped (${summary.stop})`
    summary.cities.failed.push(id)
    log('error', 'city_fail', { cityId: id, error, kept: entries.get(id)?.generatedAt ?? null })
  }
  // Not downloaded: every city needing it had failed or waits for the next run, or the run stopped.
  const attempted = new Set([...summary.extracts.ok, ...summary.extracts.failed])
  summary.extracts.skipped = order.filter((p) => !attempted.has(p))
  return { ...summary, ...geofabrik.stats, durationS: Math.round((now() - started) / 1000), ...(await memory.snapshot()) }
}

/** The process exit status for a summary (see the header). */
export function exitCode(summary, aborted = false) {
  if (aborted || summary.stop === 'aborted') return EXIT.signal
  if (summary.stop === 'breaker') return EXIT.refused
  if (summary.stop !== 'done' || summary.extracts.failed.length || summary.cities.failed.length) return EXIT.failed
  return EXIT.ok
}

// ------------------------------------------------------------------------------------------ metrics

export const PUSH_JOB = 'line-bites-refresh'

/**
 * The run's outcome in Prometheus' text format, for the Pushgateway. Every city in stations.json
 * gets the time its served file was built, 0 if never: Pushgateway values never expire, so
 * `time() - line_bites_city_generated_timestamp_seconds` keeps growing while runs fail, skip a city,
 * or never start (an init container stuck at Init:Error pushes nothing). NOTES.md, "Monitoring".
 */
export function metricsText({ cityOrder, index, exitCode, finishedAt, downloadBytes = 0 }) {
  const built = new Map((index?.cities ?? []).map((c) => [c.cityId, c]))
  const seconds = (iso) => {
    const ms = Date.parse(iso)
    return Number.isFinite(ms) ? ms / 1000 : 0
  }
  return `${[
    '# HELP line_bites_city_generated_timestamp_seconds When the served city file was built; 0 if it never was.',
    '# TYPE line_bites_city_generated_timestamp_seconds gauge',
    ...cityOrder.map((id) => `line_bites_city_generated_timestamp_seconds{city="${id}"} ${built.has(id) ? seconds(built.get(id).generatedAt) : 0}`),
    '# HELP line_bites_city_places Places in the served city file.',
    '# TYPE line_bites_city_places gauge',
    ...cityOrder.filter((id) => built.has(id)).map((id) => `line_bites_city_places{city="${id}"} ${Number(built.get(id).count) || 0}`),
    '# HELP line_bites_refresh_exit_code Exit status of the last refresh run (0 ok, 1 failed, 75 refused, 143 SIGTERM).',
    '# TYPE line_bites_refresh_exit_code gauge',
    `line_bites_refresh_exit_code ${exitCode}`,
    '# HELP line_bites_refresh_finished_timestamp_seconds When the last refresh run ended.',
    '# TYPE line_bites_refresh_finished_timestamp_seconds gauge',
    `line_bites_refresh_finished_timestamp_seconds ${finishedAt / 1000}`,
    '# HELP line_bites_refresh_download_bytes Bytes the last refresh run received from Geofabrik.',
    '# TYPE line_bites_refresh_download_bytes gauge',
    `line_bites_refresh_download_bytes ${downloadBytes}`,
  ].join('\n')}\n`
}

/** PUT replaces the job's whole group, so a city that left stations.json leaves the metrics too. */
export async function pushMetrics(baseUrl, body, { fetch = globalThis.fetch, timeoutMs = 10_000 } = {}) {
  const res = await fetch(`${baseUrl.replace(/\/+$/, '')}/metrics/job/${PUSH_JOB}`, {
    method: 'PUT',
    headers: { 'Content-Type': 'text/plain; version=0.0.4' },
    body,
    signal: AbortSignal.timeout(timeoutMs),
  })
  await res.body?.cancel()
  if (!res.ok) throw new Error(`Pushgateway: HTTP ${res.status}`)
}

async function main() {
  const log = (level, evt, fields = {}) => process.stdout.write(`${JSON.stringify({ ts: new Date().toISOString(), level, evt, ...fields })}\n`)
  const env = process.env
  const controller = new AbortController()
  for (const sig of ['SIGTERM', 'SIGINT']) {
    process.on(sig, () => {
      log('warn', 'signal', { signal: sig })
      controller.abort()
    })
  }
  const num = (name, fallback) => (env[name] === undefined || env[name] === '' ? fallback : Number(env[name]))
  const deadlineAt = Date.now() + num('DEADLINE_SECONDS', 16200) * 1000
  const dataDir = env.DATA_DIR ?? '/data'
  const stationsPath = env.STATIONS ?? fileURLToPath(new URL('./stations.json', import.meta.url))
  const stations = loadStations(await readFile(stationsPath, 'utf8'))
  const osmium = createOsmium({ bin: env.OSMIUM || 'osmium', signal: controller.signal })
  // Fails here, before the lock and any download, when osmium is missing.
  log('info', 'start', { node: process.version, osmium: await osmium.version(), cities: Object.keys(stations).length, source: GEOFABRIK_URL })
  const summary = await refresh({
    dataDir,
    workDir: env.WORK_DIR ?? '/work',
    stations,
    only: process.argv.slice(2),
    geofabrik: createGeofabrik({ log, signal: controller.signal, deadlineAt, cutoffAt: deadlineAt + TRANSFER_GRACE_MS }),
    osmium,
    log,
    allowShrink: env.ALLOW_SHRINK === '1',
    deadlineAt,
    signal: controller.signal,
  })
  const code = exitCode(summary, controller.signal.aborted)
  log(code ? 'error' : 'info', 'done', { ...summary, exitCode: code })
  process.exitCode = code
  if (env.PUSHGATEWAY_URL) {
    // Best effort: monitoring being down never fails the refresh.
    try {
      const index = JSON.parse(await readFile(join(dataDir, 'index.json'), 'utf8'))
      await pushMetrics(env.PUSHGATEWAY_URL, metricsText({ cityOrder: Object.keys(stations), index, exitCode: code, finishedAt: Date.now(), downloadBytes: summary.downloadBytes }))
      log('info', 'metrics_pushed', { job: PUSH_JOB })
    } catch (err) {
      log('warn', 'metrics_push_failed', { error: errorText(err) })
    }
  }
}

// In the pod /app/refresh-osm.mjs is a ConfigMap symlink; import.meta.url is already the resolved path.
const isMain = process.argv[1] && realpathSync(process.argv[1]) === fileURLToPath(import.meta.url)
if (isMain) {
  main().catch((err) => {
    process.stdout.write(`${JSON.stringify({ ts: new Date().toISOString(), level: 'error', evt: 'fatal', error: String(err?.stack ?? err).slice(0, 500) })}\n`)
    process.exit(1)
  })
}