Harden LVĢMC warning fetching against partial failures and type drift
This commit is contained in:
@@ -55,7 +55,12 @@ final class WarningService private (logger: Logger[IO], cache: Ref[IO, Option[(I
|
||||
fetchFresh.attempt.flatMap {
|
||||
case Right(value) => cache.set(Some(now -> value)).as(value)
|
||||
case Left(error) => stale match {
|
||||
case Some((_, value)) => logger.warn(error)("Using stale LVĢMC warning data") *> IO.pure(value)
|
||||
// Re-stamp the cache with `now` even though the value is unchanged, so a
|
||||
// failing upstream is retried at most once per cache lifetime instead of
|
||||
// on every single incoming request.
|
||||
case Some((_, value)) =>
|
||||
logger.warn(error)("Using stale LVĢMC warning data; will not retry until the cache lifetime elapses") *>
|
||||
cache.set(Some(now -> value)).as(value)
|
||||
case None => IO.raiseError(error)
|
||||
}
|
||||
}
|
||||
@@ -70,8 +75,22 @@ final class WarningService private (logger: Logger[IO], cache: Ref[IO, Option[(I
|
||||
metadataBody <- client.expect[String](metadataUri)
|
||||
metadata <- parseRecords(metadataBody, "warning metadata")
|
||||
eventIds = metadata.flatMap(record => stringValue(record, "WEATHER_WARNING_EV_ID")).distinct
|
||||
polygonRows <- eventIds.traverse(eventId => fetchPolygonRows(client, polygonsBaseUri, eventId)).map(_.flatten)
|
||||
polygons = buildPolygons(polygonRows)
|
||||
// Each event's polygons are fetched independently and concurrently; one event
|
||||
// failing (network error, malformed page, ...) must not take down every other
|
||||
// warning, so a failure is logged and treated as "no polygons for this event"
|
||||
// rather than aborting the whole refresh.
|
||||
polygonRows <- eventIds.parTraverse { eventId =>
|
||||
fetchPolygonRows(client, polygonsBaseUri, eventId).attempt.flatMap {
|
||||
case Right(rows) => IO.pure(rows)
|
||||
case Left(error) =>
|
||||
logger.warn(error)(s"Failed to fetch warning polygons for event $eventId; it will be shown without a shape or dropped") *>
|
||||
IO.pure(Vector.empty[Json])
|
||||
}
|
||||
}.map(_.flatten)
|
||||
(polygons, unresolvedOrderCount) = buildPolygons(polygonRows)
|
||||
_ <- IO.whenA(unresolvedOrderCount > 0)(
|
||||
logger.warn(s"$unresolvedOrderCount warning polygon point(s) had a missing/invalid NPK order value and were defaulted to position 0")
|
||||
)
|
||||
warnings = metadata.flatMap(record => buildWarning(record, polygons))
|
||||
.filter(_.polygons.nonEmpty)
|
||||
_ <- logger.info(s"Loaded ${warnings.size} LVĢMC warnings")
|
||||
@@ -103,7 +122,13 @@ final class WarningService private (logger: Logger[IO], cache: Ref[IO, Option[(I
|
||||
offset: Int = 0,
|
||||
accumulated: Vector[Json] = Vector.empty
|
||||
): IO[Vector[Json]] = {
|
||||
val filterValue = eventId.toLongOption.map(Json.fromLong).getOrElse(Json.fromString(eventId))
|
||||
// The polygons resource's WEATHER_WARNING_EV_ID column type isn't guaranteed, so
|
||||
// match either JSON representation via CKAN's list-filter OR semantics instead of
|
||||
// guessing one and silently returning zero rows if the guess is wrong.
|
||||
val filterValue = eventId.toLongOption match {
|
||||
case Some(numeric) => Json.arr(Json.fromString(eventId), Json.fromLong(numeric))
|
||||
case None => Json.fromString(eventId)
|
||||
}
|
||||
val filters = Json.obj("WEATHER_WARNING_EV_ID" -> filterValue).noSpaces
|
||||
val uri = baseUri
|
||||
.withQueryParam("limit", polygonPageSize.toString)
|
||||
@@ -119,31 +144,32 @@ final class WarningService private (logger: Logger[IO], cache: Ref[IO, Option[(I
|
||||
}
|
||||
}
|
||||
|
||||
// Returns the polygons grouped by event, plus a count of points whose NPK vertex-order
|
||||
// field was missing/invalid (those points are kept, defaulted to order 0, rather than
|
||||
// dropped — the caller logs the count so a garbled shape doesn't fail silently).
|
||||
private def buildPolygons(
|
||||
rows: Vector[Json]
|
||||
): Map[String, Vector[WarningPolygon]] =
|
||||
rows
|
||||
.flatMap { row =>
|
||||
): (Map[String, Vector[WarningPolygon]], Int) = {
|
||||
val points = rows.flatMap { row =>
|
||||
for {
|
||||
eventId <- stringValue(row, "WEATHER_WARNING_EV_ID")
|
||||
polygonId <- stringValue(row, "POLIGON_ID")
|
||||
lat <- doubleValue(row, "LAT")
|
||||
lon <- doubleValue(row, "LON")
|
||||
} yield ((eventId, polygonId), intValue(row, "NPK").getOrElse(0), WarningPoint(lat, lon))
|
||||
} yield (eventId, polygonId, intValue(row, "NPK"), WarningPoint(lat, lon))
|
||||
}
|
||||
.groupBy(_._1)
|
||||
.toVector
|
||||
.groupBy(_._1._1)
|
||||
.view
|
||||
.mapValues { polygonGroups =>
|
||||
polygonGroups
|
||||
.map { case ((_, polygonId), points) =>
|
||||
WarningPolygon(polygonId, points.sortBy(_._2).map(_._3))
|
||||
}
|
||||
.sortBy(_.id)
|
||||
val unresolvedOrderCount = points.count(_._3.isEmpty)
|
||||
val polygons = points
|
||||
.groupMap(point => (point._1, point._2))(point => (point._3.getOrElse(0), point._4))
|
||||
.toVector
|
||||
.map { case ((eventId, polygonId), ordered) =>
|
||||
eventId -> WarningPolygon(polygonId, ordered.sortBy(_._1).map(_._2))
|
||||
}
|
||||
.groupMap(_._1)(_._2)
|
||||
.view.mapValues(_.sortBy(_.id))
|
||||
.toMap
|
||||
(polygons, unresolvedOrderCount)
|
||||
}
|
||||
|
||||
private def buildWarning(
|
||||
record: Json,
|
||||
|
||||
Reference in New Issue
Block a user