StatefulFetchService to save last 24h of data

src/main/scala/fetch/FetchService.scala

@@ -20,13 +20,19 @@ final case class WeatherServerConfig(
     url: String,
 )
 
+trait FetchServiceTrait {
+  def fetchSingleFile(fileName: String): IO[Either[Throwable, (String, String)]]
+  def fetchInRange(from: LocalDateTime, to: LocalDateTime): IO[List[Either[Throwable, (String, String)]]]
+  def fetchFromDate(date: LocalDate): IO[List[Either[Throwable, (String, String)]]]
+}
+
 object FetchService {
   def of: IO[FetchService] = {
     Slf4jLogger.create[IO].map(logger => new FetchService(new FileNameService, logger))
   }
 }
 
-class FetchService(fileNameService: FileNameService, log: Logger[IO]) {
+class FetchService(fileNameService: FileNameService, log: Logger[IO]) extends FetchServiceTrait {
   private val weatherServerConfig: WeatherServerConfig = ConfigSource.default.load[WeatherServerConfig] match {
     case Right(config) => config
     case Left(errors) => throw new RuntimeException(s"Unable to load config: $errors")
@@ -38,7 +44,7 @@ class FetchService(fileNameService: FileNameService, log: Logger[IO]) {
   private val baseUrl: IO[Uri] = IO(Uri.unsafeFromString(weatherServerConfig.url))
 
   private def makeRequest(client: Client[IO], url: Uri): IO[Either[Throwable, (String, String)]] = {
-    val fileName = url.path.toString()
+    val fileName = url.path.toString().tail
     val request = Request[IO](Method.GET, url).withHeaders(Authorization(basicCredentials))
 
     client.expect[String](request).redeemWith(

src/main/scala/fetch/FileFetchScheduler.scala

@@ -8,7 +8,7 @@ import org.typelevel.log4cats.slf4j.Slf4jLogger
 
 
 object FileFetchScheduler {
-  def of(dbService: DBService, fetch: FetchService): IO[FileFetchScheduler] = {
+  def of(dbService: DBService, fetch: FetchServiceTrait): IO[FileFetchScheduler] = {
     Scheduler.of.flatMap { scheduler =>
       Slf4jLogger.create[IO].map {
         new FileFetchScheduler(dbService, fetch, new FileNameService(), scheduler, _)
@@ -17,7 +17,7 @@ object FileFetchScheduler {
   }
 }
 
-class FileFetchScheduler(dbService: DBService, fetch: FetchService, fileNameService: FileNameService, scheduler: Scheduler, log: Logger[IO]) {
+class FileFetchScheduler(dbService: DBService, fetch: FetchServiceTrait, fileNameService: FileNameService, scheduler: Scheduler, log: Logger[IO]) {
   def run: Stream[IO, Unit] = {
     val fetchTask = fileNameService.generateCurrentHour.flatMap(fetch.fetchSingleFile)
     scheduler.scheduleTask(fetchTask)

src/main/scala/fetch/FileNameService.scala

@@ -5,11 +5,8 @@ import cats.effect._
 import java.time.format.DateTimeFormatter
 import java.time.{Duration, LocalDate, LocalDateTime, ZoneId, ZonedDateTime}
 
-trait FileNameServiceTrait {
-  def generateCurrentHour(implicit clock: Clock[IO]): IO[String]
-}
 
-class FileNameService extends FileNameServiceTrait {
+class FileNameService {
   private val interval = Duration.ofMinutes(30)
   private val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
 
@@ -27,17 +24,30 @@ class FileNameService extends FileNameServiceTrait {
     time.plusMinutes(adjustment)
   }
 
-  override def generateCurrentHour(implicit clock: Clock[IO]): IO[String] = {
+  def generateCurrentHour(implicit clock: Clock[IO]): IO[String] = {
     clock.realTime.map { duration =>
       val instant = java.time.Instant.ofEpochMilli(duration.toMillis)
       val zonedDateTime = ZonedDateTime.ofInstant(instant, ZoneId.of("Europe/Riga"))
       val now = zonedDateTime.toLocalDateTime.withMinute(30)
       val csvName = s"${formatter.format(now)}.csv"
-//      println("println", csvName)
       csvName
     }
   }
 
+  def generateLast24Hours(implicit clock: Clock[IO]): IO[List[String]] = {
+    clock.realTime.map { duration =>
+      val instant = java.time.Instant.ofEpochMilli(duration.toMillis)
+      val zonedDateTime = ZonedDateTime.ofInstant(instant, ZoneId.of("Europe/Riga"))
+      val now = zonedDateTime.toLocalDateTime.withMinute(30)
+      val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
+
+      (0 until 24).toList.map { hour =>
+        val previousHour = now.minusHours(hour)
+        s"${formatter.format(previousHour)}.csv"
+      }
+    }
+  }
+
   // currently csv files are generated each 30 mins of hour
   def generate(startTime: LocalDateTime, endTime: LocalDateTime): List[String] = {
     val roundStartTime = roundToInterval(startTime, true)

src/main/scala/fetch/Main.scala

@@ -9,25 +9,40 @@ import org.typelevel.log4cats.slf4j.Slf4jLogger
 import java.time.LocalDateTime
 
 object Main {
+//  def run: IO[Unit] = {
+//    val from = LocalDateTime.of(2023, 4, 28, 10, 0)
+//    val to = LocalDateTime.of(2023, 4, 28, 13, 30)
+//    for {
+//      log <- Slf4jLogger.create[IO]
+//      fetch <- FetchService.of
+//      //      fetchResultEither <- fetch.fetchFromDate(LocalDate.of(2023, 4, 28)).attempt
+//      fetchResultEither <- fetch.fetchInRange(from, to).attempt
+//      fetchServiceError = fetchResultEither.left.toOption.map(e => s"FetchServiceError: ${e.getMessage}").toList
+//      fetchResult = fetchResultEither.getOrElse(List.empty)
+//      (fetchErrors, successDownloads) = fetchResult.partitionMap(identity)
+//      dbService <- DBService.of
+//      saveResults <- successDownloads.traverse { case (name, content) => dbService.save(name, content) }
+//      (saveErrors, successSaves) = saveResults.partitionMap(identity)
+//      successes = successDownloads.map(s => s"fetched: ${s._1}") ++ successSaves.map(s => s"saved: $s")
+//      errors = fetchServiceError ++ fetchErrors.map(e => s"FetchError: ${e.getMessage}") ++ saveErrors.map(e => s"SaveError: ${e.getMessage}")
+//      _ <- log.info(s"errors: $errors")
+//      _ <- log.info(s"successes: $successes")
+//    } yield (successes, errors)
+//  }
+
   def run: IO[Unit] = {
-    val from = LocalDateTime.of(2023, 4, 28, 10, 0)
-    val to = LocalDateTime.of(2023, 4, 28, 13, 30)
     for {
-      log <- Slf4jLogger.create[IO]
       fetch <- FetchService.of
-//      fetchResultEither <- fetch.fetchFromDate(LocalDate.of(2023, 4, 28)).attempt
-      fetchResultEither <- fetch.fetchInRange(from, to).attempt
+      statefulFetch <- StatefulFetchService.of(fetch)
+      fetchResultEither <- statefulFetch.fetchSingleFile("20230524_0030.csv").attempt
+      fetchResultEither <- statefulFetch.fetchSingleFile("20230522_0130.csv").attempt
       fetchServiceError = fetchResultEither.left.toOption.map(e => s"FetchServiceError: ${e.getMessage}").toList
-      fetchResult = fetchResultEither.getOrElse(List.empty)
-      (fetchErrors, successDownloads) = fetchResult.partitionMap(identity)
-      dbService <- DBService.of
-      saveResults <- successDownloads.traverse { case (name, content) => dbService.save(name, content) }
-      (saveErrors, successSaves) = saveResults.partitionMap(identity)
-      successes = successDownloads.map(s => s"fetched: ${s._1}") ++ successSaves.map(s => s"saved: $s")
-      errors = fetchServiceError ++ fetchErrors.map(e => s"FetchError: ${e.getMessage}") ++ saveErrors.map(e => s"SaveError: ${e.getMessage}")
-      _ <- log.info(s"errors: $errors")
-      _ <- log.info(s"successes: $successes")
-    } yield (successes, errors)
+      fetchResult = fetchResultEither.flatMap(res => res.flatMap(aaa => {
+        println(s"fffffff: ${aaa._1}")
+        Right(aaa._1)
+      }))
+//      _ = println(s"${fetchResult.map()}")
+    } yield ()
   }
 
   def main(args: Array[String]): Unit = {

src/main/scala/fetch/StatefulFetchService.scala

@@ -0,0 +1,59 @@
+package fetch
+
+import cats.effect.{IO, Ref}
+import cats.implicits.toTraverseOps
+import org.typelevel.log4cats.Logger
+import org.typelevel.log4cats.slf4j.Slf4jLogger
+
+import java.time.{LocalDate, LocalDateTime}
+
+object StatefulFetchService {
+  def of(fetchService: FetchService): IO[StatefulFetchService] = {
+    Slf4jLogger.create[IO].map(logger => new StatefulFetchService(fetchService, new FileNameService(), logger))
+  }
+}
+
+class StatefulFetchService(fetchService: FetchService, fileNameService: FileNameService, log: Logger[IO]) extends FetchServiceTrait {
+  private val state: Ref[IO, Map[String, String]] = Ref.unsafe(Map.empty)
+
+  private def logState: IO[Unit] = {
+    state.get.flatMap(currentState => log.info(s"State keys: ${currentState.keys}"))
+  }
+
+  private def updateState(fileName: String, content: String): IO[Unit] = {
+    for {
+      last24Hours <- fileNameService.generateLast24Hours
+      _ <- state.update(st => (st + (fileName -> content)).filterKeys(last24Hours.contains).toMap)
+      _ <- logState
+    } yield ()
+  }
+
+  def fetchSingleFile(fileName: String): IO[Either[Throwable, (String, String)]] = {
+    fetchService.fetchSingleFile(fileName).flatMap {
+      case Right((fileName, content)) =>
+        updateState(fileName, content).as(Right((fileName, content)))
+      case e@Left(_) => IO(e)
+    }
+  }
+
+  def fetchInRange(from: LocalDateTime, to: LocalDateTime): IO[List[Either[Throwable, (String, String)]]] = {
+    fetchService.fetchInRange(from, to).flatMap { results =>
+      val successfulResults = results.collect { case Right(data) => data }
+      successfulResults.traverse { case (name, content) => updateState(name, content) }.as(results)
+    }
+  }
+
+  def fetchFromDate(date: LocalDate): IO[List[Either[Throwable, (String, String)]]] = {
+    fetchService.fetchFromDate(date).flatMap { results =>
+      val successfulResults = results.collect { case Right(data) => data }
+      successfulResults.traverse { case (name, content) => updateState(name, content) }.as(results)
+    }
+  }
+}
+
+
+
+
+
+
+

src/main/scala/server/Main.scala

@@ -2,16 +2,17 @@ package server
 import cats.effect._
 import cats.implicits.catsSyntaxTuple2Parallel
 import db.DBService
-import fetch.{FetchService, FileFetchScheduler}
+import fetch.{FetchService, FileFetchScheduler, StatefulFetchService}
 
 object Main extends IOApp {
   def run(args: List[String]): IO[ExitCode] = {
     for {
       dbService <- DBService.of
       fetch <- FetchService.of
-      fileFetchScheduler <- FileFetchScheduler.of(dbService, fetch)
+      statefulFetch <- StatefulFetchService.of(fetch)
+      fileFetchScheduler <- FileFetchScheduler.of(dbService, statefulFetch)
       schedulerTask = fileFetchScheduler.run.compile.drain
-      server <- Server.of(dbService, fetch)
+      server <- Server.of(dbService, statefulFetch)
       serverTask = server.run
       exitCode <- (serverTask, schedulerTask).parMapN((_, _) => ExitCode.Success)
     } yield exitCode

src/main/scala/server/Server.scala

@@ -4,7 +4,7 @@ import cats.effect._
 import cats.implicits.toTraverseOps
 import com.comcast.ip4s.IpLiteralSyntax
 import db.DBService
-import fetch.FetchService
+import fetch.FetchServiceTrait
 import parse.{Parser, WeatherData}
 import server.ValidateRoutes.{AggKey, CityList, DateTimeRange, ValidDate}
 import io.circe.{Json, Printer}
@@ -26,14 +26,14 @@ import scala.concurrent.duration.DurationInt
 
 
 object Server {
-  def of(dbService: DBService, fetch: FetchService): IO[Server] = {
+  def of(dbService: DBService, fetch: FetchServiceTrait): IO[Server] = {
     Slf4jLogger.create[IO].map {
       new Server(dbService, fetch, _)
     }
   }
 }
 
-class Server(dbService: DBService, fetch: FetchService, log: Logger[IO]) {
+class Server(dbService: DBService, fetch: FetchServiceTrait, log: Logger[IO]) {
 
   // Define the extension method `pretty` for Json
   implicit class JsonPrettyPrinter(json: Json) {

src/test/scala/fetch/FileNameServiceSpec.scala

@@ -1,10 +1,16 @@
+import cats.Applicative
+import cats.effect.unsafe.implicits.global
 import fetch.FileNameService
 import org.scalatest.funsuite.AnyFunSuite
+import cats.effect.{Clock, IO}
 
+import java.util.concurrent.TimeUnit
 import java.time.LocalDateTime
+import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.duration._
+
 
 class FileNameServiceSpec extends AnyFunSuite {
-
   test("FileNameService.generate returns correct file names") {
     val startTime = LocalDateTime.of(2023, 5, 17, 0, 0)
     val endTime = LocalDateTime.of(2023, 5, 17, 4, 0)
@@ -32,4 +38,33 @@ class FileNameServiceSpec extends AnyFunSuite {
 
     assert(actualFileNames == expectedFileNames)
   }
+
+  val fixedClock: Clock[IO] = new Clock[IO] {
+    val timeInMillis = 1618862447000L
+
+    override def realTime: IO[FiniteDuration] =
+      IO.pure(Duration.fromNanos(TimeUnit.MILLISECONDS.toNanos(timeInMillis)))
+
+    override def applicative: Applicative[IO] =
+      Applicative.apply
+
+    override def monotonic: IO[FiniteDuration] =
+      IO.pure(Duration.fromNanos(TimeUnit.MILLISECONDS.toNanos(timeInMillis)))
+
+    override def timed[A](fa: IO[A]): IO[(FiniteDuration, A)] =
+      fa.map(a => (Duration.Zero, a))
+  }
+
+  test("FileNameService.generateLast24Hours returns correct file names for a fixed time") {
+    val fileNameService = new FileNameService()
+    val filenames = fileNameService.generateLast24Hours(fixedClock).unsafeRunSync()
+
+    assert(filenames.size == 24)
+
+    val expectedFileNames = (0 to 23).toList
+      .map(hour => if (hour < 10) "0" + hour else "" + hour)
+      .map(str => s"20210419_${str}30.csv")
+
+    assert(filenames.sorted == expectedFileNames.sorted)
+  }
 }