Harden LVĢMC warning fetching against partial failures and type drift

src/main/scala/fetch/warnings/WarningService.scala

@@ -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)
+        // Each event's polygons are fetched independently and concurrently; one event
-        polygons = buildPolygons(polygonRows)
+        // 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]] =
+  ): (Map[String, Vector[WarningPolygon]], Int) = {
-    rows
+    val points = rows.flatMap { row =>
-      .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)
+    val unresolvedOrderCount = points.count(_._3.isEmpty)
-      .toVector
+    val polygons = points
-      .groupBy(_._1._1)
+      .groupMap(point => (point._1, point._2))(point => (point._3.getOrElse(0), point._4))
-      .view
-      .mapValues { polygonGroups =>
-        polygonGroups
-          .map { case ((_, polygonId), points) =>
-            WarningPolygon(polygonId, points.sortBy(_._2).map(_._3))
-          }
-          .sortBy(_.id)
       .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,