WeatherTool/src/main/scala/fetch/lvgmc/WaterTemperatureService.scala

package fetch.lvgmc

import cats.effect.{IO, Ref}
import cats.syntax.all._
import io.circe.Json
import io.circe.parser
import org.http4s.Uri
import org.http4s.ember.client.EmberClientBuilder
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger

import java.time.{Duration, Instant}

final case class WaterTemperatureReading(area: String, observedAt: String, value: Option[Double])
final case class WaterTemperatureResponse(source: String, readings: Vector[WaterTemperatureReading])

object WaterTemperatureService {
  def of: IO[WaterTemperatureService] =
    for {
      logger <- Slf4jLogger.create[IO]
      cache <- Ref.of[IO, Option[(Instant, WaterTemperatureResponse)]](None)
    } yield new WaterTemperatureService(logger, cache)
}

// Real-time only, by design: the Ūdens workspace has never persisted history
// (its ranges are manual editorial inputs), and this follows the same
// pattern — an in-memory cache, no database table, mirroring
// fetch.warnings.WarningService rather than the station-observation path.
final class WaterTemperatureService private (logger: Logger[IO], cache: Ref[IO, Option[(Instant, WaterTemperatureResponse)]]) {
  private val cacheLifetime = Duration.ofMinutes(15)

  private val observationsUrl = sys.env.getOrElse(
    "LVGMC_OPENDATA_HYDRO_URL",
    "https://data.gov.lv/dati/api/3/action/datastore_search?resource_id=de5f06e9-6f44-497d-8ec2-72a2483608e8"
  )

  // One representative real station per named water area on the Ūdens
  // production template, verified 2026-08-23 against LVĢMC's open
  // hydrological data. Jūra/Līcis are coastal sea-temperature stations
  // (SEDUT); the four historical regions use inland river/lake stations
  // (WTEMD). Approximate by design — see docs/UPDATE_ROADMAP.md.
  private val areaStations: List[(String, String, String)] = List(
    ("Jūra", "RIVE99PA", "SEDUT"),
    ("Līcis", "RISE99MS", "SEDUT"),
    ("Kurzeme", "HD073613", "WTEMD"),
    ("Zemgale", "HD073801", "WTEMD"),
    ("Vidzeme", "HD073062", "WTEMD"),
    ("Latgale", "HD072971", "WTEMD"), // Ludza — Daugavpils's station (HD073141) only reports water level (LIMEN), not temperature
  )

  def fetchWaterTemperatures: IO[WaterTemperatureResponse] =
    for {
      now <- IO.delay(Instant.now)
      cached <- cache.get
      response <- cached match {
        case Some((storedAt, value)) if Duration.between(storedAt, now).compareTo(cacheLifetime) < 0 =>
          IO.pure(value)
        case stale =>
          fetchFresh.attempt.flatMap {
            case Right(value) => cache.set(Some(now -> value)).as(value)
            case Left(error) => stale match {
              // Re-stamp with `now` even on failure so a dead upstream is
              // retried at most once per cache lifetime, not on every request
              // (see the identical fix in WarningService).
              case Some((_, value)) =>
                logger.warn(error)("Using stale water temperature data; will not retry until the cache lifetime elapses") *>
                  cache.set(Some(now -> value)).as(value)
              case None => IO.raiseError(error)
            }
          }
      }
    } yield response

  private def fetchFresh: IO[WaterTemperatureResponse] =
    EmberClientBuilder.default[IO].build.use { client =>
      for {
        baseUri <- Uri.fromString(observationsUrl).liftTo[IO]
        stationIds = areaStations.map(_._2).distinct
        abbreviations = areaStations.map(_._3).distinct
        filters = Json.obj(
          "STATION_ID" -> Json.arr(stationIds.map(Json.fromString): _*),
          "ABBREVIATION" -> Json.arr(abbreviations.map(Json.fromString): _*),
        ).noSpaces
        // The underlying resource is LVĢMC's rolling 48h operational window,
        // so this narrow a filter (6 stations, 2 parameters) never needs
        // pagination the way the larger station/warning fetches do.
        uri = baseUri.withQueryParam("limit", "2000").withQueryParam("filters", filters)
        body <- client.expect[String](uri)
        rows <- parseRecords(body)
        readings = areaStations.map { case (area, stationId, abbreviation) =>
          latestFor(rows, stationId, abbreviation) match {
            case Some((observedAt, value)) => WaterTemperatureReading(area, observedAt, Some(value))
            case None => WaterTemperatureReading(area, "", None)
          }
        }
        _ <- logger.info(s"Loaded water temperatures for ${readings.count(_.value.isDefined)} of ${readings.size} areas")
      } yield WaterTemperatureResponse("LVĢMC", readings.toVector)
    }

  private def latestFor(rows: Vector[Json], stationId: String, abbreviation: String): Option[(String, Double)] =
    rows.flatMap { row =>
      val cursor = row.hcursor
      for {
        sid <- cursor.downField("STATION_ID").as[String].toOption if sid == stationId
        ab <- cursor.downField("ABBREVIATION").as[String].toOption if ab == abbreviation
        datetime <- cursor.downField("DATETIME").as[String].toOption
        value <- cursor.downField("VALUE").as[Double].toOption
      } yield datetime -> value
    }.sortBy(_._1)(Ordering[String].reverse).headOption

  private def parseRecords(body: String): IO[Vector[Json]] =
    IO.fromEither(
      parser.parse(body)
        .leftMap(error => new RuntimeException(s"Unable to parse water temperature data: ${error.message}"))
        .flatMap { json =>
          json.hcursor.downField("result").downField("records").as[Vector[Json]]
            .leftMap(error => new RuntimeException(s"Invalid water temperature response: ${error.message}"))
        }
    )
}