Update Flyway and Elastic4s

build.sbt

@@ -67,7 +67,7 @@ lazy val infra = project
     libraryDependencies ++= Seq(
       "com.sksamuel.elastic4s" %% "elastic4s-client-esjava" % V.elastic4sVersion,
       "org.json4s" %% "json4s-native" % V.json4s,
-      "org.flywaydb" % "flyway-core" % "8.5.13", // for database migration
+      "org.flywaydb" % "flyway-core" % "9.8.3", // for database migration
       "com.typesafe.akka" %% "akka-stream" % V.akkaVersion,
       "com.typesafe.akka" %% "akka-http" % V.akkaHttpVersion,
       "de.heikoseeberger" %% "akka-http-circe" % "1.39.2",
@@ -244,7 +244,7 @@ lazy val V = new {
   val playJsonVersion = "2.9.3"
   val akkaVersion = "2.6.18"
   val akkaHttpVersion = "10.2.10"
-  val elastic4sVersion = "8.4.4"
+  val elastic4sVersion = "8.5.0"
   val nscalaTimeVersion = "2.32.0"
   val scalatest = "3.2.14"
   val circeVersion = "0.14.3"

modules/infra/src/main/scala/scaladex/infra/SqlDatabase.scala

@@ -34,8 +34,10 @@ import scaladex.infra.sql.ReleaseDependenciesTable
 import scaladex.infra.sql.ReleaseTable
 import scaladex.infra.sql.UserSessionsTable
 
-class SqlDatabase(datasource: HikariDataSource, xa: doobie.Transactor[IO]) extends SchedulerDatabase with LazyLogging {
-  private val flyway = DoobieUtils.flyway(datasource)
+class SqlDatabase(datasource: HikariDataSource, xa: doobie.Transactor[IO], testMode: Boolean = false)
+    extends SchedulerDatabase
+    with LazyLogging {
+  private val flyway = DoobieUtils.flyway(datasource, testMode)
   def migrate: IO[Unit] = IO(flyway.migrate())
   def dropTables: IO[Unit] = IO(flyway.clean())
 

modules/infra/src/main/scala/scaladex/infra/migrations/FlywayMigration.scala

@@ -0,0 +1,29 @@
+package scaladex.infra.migrations
+
+import scala.concurrent.ExecutionContext
+
+import cats.effect.ContextShift
+import cats.effect.IO
+import com.typesafe.scalalogging.LazyLogging
+import doobie.implicits._
+import doobie.util.transactor.Transactor
+import org.flywaydb.core.api.migration.BaseJavaMigration
+import org.flywaydb.core.api.migration.Context
+import scaladex.infra.config.PostgreSQLConfig
+
+abstract class FlywayMigration extends BaseJavaMigration with LazyLogging {
+  def migrationIO: IO[Unit]
+
+  def run[A](v: doobie.ConnectionIO[A]): IO[A] =
+    v.transact(FlywayMigration.transactor)
+
+  override def migrate(context: Context): Unit =
+    migrationIO.unsafeRunSync()
+}
+
+object FlywayMigration {
+  private implicit val cs: ContextShift[IO] = IO.contextShift(ExecutionContext.global)
+  private val config = PostgreSQLConfig.load().get
+  private val transactor: Transactor.Aux[IO, Unit] =
+    Transactor.fromDriverManager[IO](config.driver, config.url, config.user, config.pass.decode)
+}

modules/infra/src/main/scala/scaladex/infra/migrations/V11_2__add_data_to_the_new_tables.scala

@@ -1,24 +1,13 @@
 package scaladex.infra.migrations
 
-import com.typesafe.scalalogging.LazyLogging
-import org.flywaydb.core.api.migration.BaseJavaMigration
-import org.flywaydb.core.api.migration.Context
+import cats.effect.IO
 import scaladex.infra.sql.ArtifactTable
 import scaladex.infra.sql.ReleaseTable
 
-class V11_2__add_data_to_the_new_tables extends BaseJavaMigration with ScaladexBaseMigration with LazyLogging {
-  override def migrate(context: Context): Unit =
-    try {
-      (for {
-        releases <- run(xa)(ArtifactTable.getReleasesFromArtifacts.to[Seq])
-        _ <- run(xa)(ReleaseTable.insertIfNotExists.updateMany(releases))
-      } yield ())
-        .unsafeRunSync()
-
-    } catch {
-      case e: Throwable =>
-        logger.info("failed to migrate the database")
-        throw new Exception(s"failed to migrate the database because of ${e.getMessage}")
-    }
-
+class V11_2__add_data_to_the_new_tables extends FlywayMigration {
+  override def migrationIO: IO[Unit] =
+    for {
+      releases <- run(ArtifactTable.getReleasesFromArtifacts.to[Seq])
+      _ <- run(ReleaseTable.insertIfNotExists.updateMany(releases))
+    } yield ()
 }

modules/infra/src/main/scala/scaladex/infra/migrations/V13_2__update_new_fields_in_artifacts.scala

@@ -1,36 +1,26 @@
 package scaladex.infra.migrations
+import cats.effect.IO
 import cats.implicits._
-import com.typesafe.scalalogging.LazyLogging
 import doobie.Query0
 import doobie.util.update.Update
-import org.flywaydb.core.api.migration.BaseJavaMigration
-import org.flywaydb.core.api.migration.Context
 import scaladex.core.model.Artifact
 import scaladex.core.model.Artifact.MavenReference
 import scaladex.infra.sql.DoobieUtils.Mappings._
 import scaladex.infra.sql.DoobieUtils.selectRequest
 import scaladex.infra.sql.DoobieUtils.updateRequest
 
-class V13_2__update_new_fields_in_artifacts extends BaseJavaMigration with ScaladexBaseMigration with LazyLogging {
-  override def migrate(context: Context): Unit =
-    try {
-      (for {
-        oldArtifacts <- run(xa)(selectArtifact.to[Seq])
-        groupedArtifacts = oldArtifacts.grouped(10000).toSeq
-        _ <- groupedArtifacts
-          .map(artifacts => run(xa)(updateNewFields.updateMany(artifacts.map(update))))
-          .sequence
-      } yield ())
-        .unsafeRunSync()
+class V13_2__update_new_fields_in_artifacts extends FlywayMigration {
+  override def migrationIO: IO[Unit] =
+    for {
+      oldArtifacts <- run(selectArtifact.to[Seq])
+      groupedArtifacts = oldArtifacts.grouped(10000).toSeq
+      _ <- groupedArtifacts
+        .map(artifacts => run(updateNewFields.updateMany(artifacts.map(update))))
+        .sequence
+    } yield ()
 
-    } catch {
-      case e: Throwable =>
-        logger.info("failed to migrate the database")
-        throw new Exception(s"failed to migrate the database because of ${e.getMessage}")
-    }
-
-  val selectArtifact: Query0[Artifact] = selectRequest("artifacts", Seq("*"))
-  val updateNewFields: Update[(Boolean, Boolean, MavenReference)] =
+  private val selectArtifact: Query0[Artifact] = selectRequest("artifacts", Seq("*"))
+  private val updateNewFields: Update[(Boolean, Boolean, MavenReference)] =
     updateRequest("artifacts", Seq("is_semantic", "is_prerelease"), Seq("group_id", "artifact_id", "version"))
 
   private def update(artifact: Artifact): (Boolean, Boolean, MavenReference) =

modules/infra/src/main/scala/scaladex/infra/migrations/V17__add_mill_platform.scala

@@ -1,27 +1,23 @@
 package scaladex.infra.migrations
 
-import com.typesafe.scalalogging.LazyLogging
+import cats.effect.IO
 import doobie._
-import org.flywaydb.core.api.migration.BaseJavaMigration
-import org.flywaydb.core.api.migration.Context
 import scaladex.core.model._
 import scaladex.infra.sql.DoobieUtils.Mappings._
 import scaladex.infra.sql.DoobieUtils._
 
-class V17__add_mill_platform extends BaseJavaMigration with ScaladexBaseMigration with LazyLogging {
-  override def migrate(context: Context): Unit = {
-    val migrateIO = for {
-      oldArtifacts <- run(xa)(selectArtifact.to[Seq])
+class V17__add_mill_platform extends FlywayMigration {
+  override def migrationIO: IO[Unit] =
+    for {
+      oldArtifacts <- run(selectArtifact.to[Seq])
       newArtifacts = oldArtifacts.map { a =>
         val newId = Artifact.ArtifactId.parse(a.artifactId).get
         a.copy(artifactName = newId.name, platform = newId.binaryVersion.platform)
       }
       (toUpdate, toDelete) = newArtifacts.partition(a => isValidMillPlugin(a))
-      _ <- run(xa)(updateNewFields.updateMany(toUpdate.map(update)))
-      _ <- run(xa)(delete.updateMany(toDelete.map(_.mavenReference)))
+      _ <- run(updateNewFields.updateMany(toUpdate.map(update)))
+      _ <- run(delete.updateMany(toDelete.map(_.mavenReference)))
     } yield ()
-    migrateIO.unsafeRunSync()
-  }
 
   val selectArtifact: Query0[Artifact] =
     selectRequest("artifacts", Seq("*"), where = Seq("artifact_name LIKE '%_mill0_%'"))

modules/infra/src/main/scala/scaladex/infra/migrations/V7_2__edit_platform_and_language.scala

@@ -2,40 +2,30 @@ package scaladex.infra.migrations
 
 import java.time.Instant
 
+import cats.effect.IO
 import cats.implicits._
-import com.typesafe.scalalogging.LazyLogging
 import doobie.Query0
 import doobie.implicits._
 import doobie.util.update.Update
-import org.flywaydb.core.api.migration.BaseJavaMigration
-import org.flywaydb.core.api.migration.Context
 import scaladex.core.model.Artifact.MavenReference
 import scaladex.core.model._
 import scaladex.infra.sql.DoobieUtils.Mappings._
 import scaladex.infra.sql.DoobieUtils.selectRequest
 import scaladex.infra.sql.DoobieUtils.updateRequest
 
-class V7_2__edit_platform_and_language extends BaseJavaMigration with ScaladexBaseMigration with LazyLogging {
+class V7_2__edit_platform_and_language extends FlywayMigration {
 
   import V7_2__edit_platform_and_language._
 
-  override def migrate(context: Context): Unit =
-    try {
-      (for {
-        oldArtifacts <- run(xa)(selectArtifact.to[Seq])
-        groupedArtifacts = oldArtifacts.grouped(10000).toSeq
-        _ <- groupedArtifacts
-          .map(artifacts => run(xa)(updatePlatformAndLanguage.updateMany(artifacts.map(_.update))))
-          .sequence
-        _ <- run(xa)(sql"ALTER TABLE artifacts DROP COLUMN binary_version".update.run)
-      } yield ())
-        .unsafeRunSync()
-
-    } catch {
-      case e: Throwable =>
-        logger.info("failed to migrate the database")
-        throw new Exception(s"failed to migrate the database because of ${e.getMessage}")
-    }
+  override def migrationIO: IO[Unit] =
+    for {
+      oldArtifacts <- run(selectArtifact.to[Seq])
+      groupedArtifacts = oldArtifacts.grouped(10000).toSeq
+      _ <- groupedArtifacts
+        .map(artifacts => run(updatePlatformAndLanguage.updateMany(artifacts.map(_.update))))
+        .sequence
+      _ <- run(sql"ALTER TABLE artifacts DROP COLUMN binary_version".update.run)
+    } yield ()
 
   val selectArtifact: Query0[OldArtifact] = selectRequest("artifacts", Seq("*"))
 

modules/infra/src/main/scala/scaladex/infra/migrations/V9__fix_platform_and_language.scala

@@ -1,34 +1,24 @@
 package scaladex.infra.migrations
-import com.typesafe.scalalogging.LazyLogging
+import cats.effect.IO
 import doobie.Query0
 import doobie.util.update.Update
-import org.flywaydb.core.api.migration.BaseJavaMigration
-import org.flywaydb.core.api.migration.Context
 import scaladex.core.model.Artifact.MavenReference
 import scaladex.core.model._
 import scaladex.infra.sql.DoobieUtils.Mappings._
 import scaladex.infra.sql.DoobieUtils.selectRequest
 import scaladex.infra.sql.DoobieUtils.updateRequest
 
-class V9__fix_platform_and_language extends BaseJavaMigration with ScaladexBaseMigration with LazyLogging {
-  override def migrate(context: Context): Unit =
-    try {
-      (for {
-        artifactToFix <- run(xa)(selectArtifact.to[Seq])
-        artifactToFixWithIds = artifactToFix.flatMap(a => Artifact.ArtifactId.parse(a.artifactId).map(a -> _))
-        _ <- run(xa) {
-          updatePlatformAndLanguage.updateMany(artifactToFixWithIds.map {
-            case (artifact, id) => (id.binaryVersion.platform, id.binaryVersion.language, artifact.mavenReference)
-          })
-        }
-      } yield ())
-        .unsafeRunSync()
-
-    } catch {
-      case e: Throwable =>
-        logger.info("failed to migrate the database")
-        throw new Exception(s"failed to migrate the database because of ${e.getMessage}")
-    }
+class V9__fix_platform_and_language extends FlywayMigration {
+  override def migrationIO: IO[Unit] =
+    for {
+      artifactToFix <- run(selectArtifact.to[Seq])
+      artifactToFixWithIds = artifactToFix.flatMap(a => Artifact.ArtifactId.parse(a.artifactId).map(a -> _))
+      _ <- run {
+        updatePlatformAndLanguage.updateMany(artifactToFixWithIds.map {
+          case (artifact, id) => (id.binaryVersion.platform, id.binaryVersion.language, artifact.mavenReference)
+        })
+      }
+    } yield ()
 
   val selectArtifact: Query0[Artifact] =
     selectRequest("artifacts", Seq("*"), where = Seq("language_version = 'Java'", "version ~ '^[^.]*$'"))

modules/infra/src/main/scala/scaladex/infra/sql/DoobieUtils.scala

@@ -43,14 +43,11 @@ object DoobieUtils {
   private implicit val cs: ContextShift[IO] =
     IO.contextShift(ExecutionContext.global)
 
-  def flyway(conf: PostgreSQLConfig): Flyway = {
-    val datasource = getHikariDataSource(conf)
-    flyway(datasource)
-  }
-  def flyway(datasource: HikariDataSource): Flyway =
+  def flyway(datasource: HikariDataSource, testMode: Boolean): Flyway =
     Flyway
       .configure()
       .dataSource(datasource)
+      .cleanDisabled(!testMode)
       .locations("migrations", "scaladex/infra/migrations")
       .load()