Universal scheduler, cleanup task for scheduler

src/main/scala/app/Main.scala

@@ -1,7 +1,8 @@
 package app
 
 import cats.effect._
-import cats.implicits.catsSyntaxTuple2Parallel
+import cats.implicits.catsSyntaxTuple3Parallel
+import data.DataService
 import db.{DBConnection, PostgresService}
 import fetch.{FetchService, FileFetchScheduler}
 import server.Server
@@ -17,10 +18,25 @@ object Main extends IOApp {
       fileFetchScheduler <- FileFetchScheduler.of(postgresService, fetch)
       schedulerTask = fileFetchScheduler.run.compile.drain
 
+
+//      scheduler <- fetchDMI.Scheduler.of("Grib", List(3, 27, 39, 51))
+//      simpleTask = IO.delay {
+//        val timeNow = LocalTime.now().format(DateTimeFormatter.ofPattern("HH:mm"))
+//        List(s"Task executed at $timeNow")
+//      }
+//      simpleScheduler = scheduler.scheduleTask(simpleTask).compile.drain
+
+
+      scheduler <- fetchDMI.Scheduler.of("Cleanup", List(1))
+      cleanupTask = DataService.deleteOldForecasts()
+      cleanupScheduler = scheduler.scheduleTask(cleanupTask).compile.drain
+
+
+
       server <- Server.of(postgresService, fetch)
       serverTask = server.run
 
-      exitCode <- (serverTask, schedulerTask).parMapN((_, _) => ExitCode.Success)
+      exitCode <- (serverTask, schedulerTask, cleanupScheduler).parMapN((_, _, _) => ExitCode.Success)
     } yield exitCode
   }
 }

src/main/scala/fetch/Scheduler.scala

@@ -34,7 +34,7 @@ class Scheduler(log: Logger[IO]) {
 
   def scheduleTask(task: IO[Either[Throwable, (String, String)]]): Stream[IO, Either[Throwable, (String, String)]] = {
     Stream.eval(durationToNextHalfHour).flatMap { delay => {
-      Stream.eval(log.info(s"Scheduler started with delay: ${delay.toMinutes} min")) *>
+      Stream.eval(log.info(s"CSV scheduler started with delay: ${delay.toMinutes} min")) *>
       (Stream.sleep[IO](delay) ++ Stream.awakeEvery[IO](1.hour))
         .evalMap(_ => task)
     }}

src/main/scala/fetchDMI/Scheduler.scala

@@ -0,0 +1,42 @@
+package fetchDMI
+
+import cats.effect._
+import cats.implicits.catsSyntaxApply
+import fs2.Stream
+import org.typelevel.log4cats.Logger
+import org.typelevel.log4cats.slf4j.Slf4jLogger
+
+import java.time.{Duration, LocalTime}
+import scala.concurrent.duration._
+
+
+object Scheduler {
+  def of(name: String, minutes: List[Int]): IO[Scheduler] = {
+    Slf4jLogger.create[IO].map(logger => new Scheduler(name, minutes, logger))
+  }
+}
+
+class Scheduler(name: String, minutes: List[Int], log: Logger[IO]) {
+  private def durationToNext(targetMinute: Int)(implicit clock: Clock[IO]): IO[FiniteDuration] = {
+    clock.realTime.map { duration =>
+      val now = LocalTime.ofSecondOfDay((duration.toMillis / 1000) % (24 * 60 * 60))
+      val nextTime =
+        if (now.getMinute < targetMinute) now.withMinute(targetMinute)
+        else now.plusHours(1).withMinute(targetMinute)
+      val durationToNext = Duration.between(now, nextTime)
+      FiniteDuration(durationToNext.toMillis, MILLISECONDS)
+    }
+  }
+
+  def scheduleTask(task: IO[List[String]]): Stream[IO, List[String]] = {
+    def streamForMinute(minute: Int): Stream[IO, List[String]] = {
+      Stream.eval(durationToNext(minute)).flatMap { delay =>
+      Stream.eval(log.info(s"$name scheduler in: ${delay.toMinutes} min")) *>
+        (Stream.sleep[IO](delay) ++ Stream.awakeEvery[IO](1.hour))
+          .evalMap(_ => task)
+      }
+    }
+
+    minutes.map(streamForMinute).reduce(_ merge _)
+  }
+}