Refactor scala base db migrations. Add marker trait for task args

modules/store/src/main/scala/db/migration/common/JsonCodecs.scala

@@ -0,0 +1,14 @@
+package db.migration.common
+
+import emil.MailAddress
+import emil.javamail.syntax._
+import io.circe.{Decoder, Encoder}
+
+trait JsonCodecs {
+
+  implicit val jsonEncoder: Encoder[MailAddress] =
+    Encoder.encodeString.contramap(_.asUnicodeString)
+  implicit val jsonDecoder: Decoder[MailAddress] =
+    Decoder.decodeString.emap(MailAddress.parse)
+
+}

modules/store/src/main/scala/db/migration/common/MigrateDueItemTasks.scala

@@ -1,21 +1,13 @@
-/*
- * Copyright 2020 Eike K. & Contributors
- *
- * SPDX-License-Identifier: AGPL-3.0-or-later
- */
-
-package db.migration
+package db.migration.common
 
 import cats.data.{NonEmptyList, OptionT}
 import cats.effect.{IO, Sync}
 import cats.implicits._
-
 import docspell.common._
 import docspell.common.syntax.StringSyntax._
 import docspell.notification.api._
 import docspell.store.queries.QLogin
 import docspell.store.records._
-
 import db.migration.data.{
   PeriodicDueItemsArgs => PeriodicDueItemsArgsLegacy,
   PeriodicQueryArgs => PeriodicQueryArgsLegacy,
@@ -23,31 +15,15 @@ import db.migration.data.{
 }
 import doobie._
 import doobie.implicits._
-import doobie.util.transactor.Strategy
-import emil.MailAddress
-import emil.javamail.syntax._
 import io.circe.syntax._
-import io.circe.{Decoder, Encoder}
 import org.flywaydb.core.api.migration.Context
 
-trait MigrationTasks {
+object MigrateDueItemTasks extends TransactorSupport with JsonCodecs {
 
-  def logger: org.log4s.Logger
+  val logger = docspell.logging.getLogger[IO]
 
-  implicit val jsonEncoder: Encoder[MailAddress] =
-    Encoder.encodeString.contramap(_.asUnicodeString)
-  implicit val jsonDecoder: Decoder[MailAddress] =
-    Decoder.decodeString.emap(MailAddress.parse)
-
-  def migrateDueItemTasks: ConnectionIO[Unit] =
-    for {
-      tasks <- RPeriodicTask.findByTask(NotifyDueItemsArgs.taskName)
-      _ <- Sync[ConnectionIO].delay(
-        logger.info(s"Starting to migrate ${tasks.size} user tasks")
-      )
-      _ <- tasks.traverse(migrateDueItemTask1)
-      _ <- RPeriodicTask.setEnabledByTask(NotifyDueItemsArgs.taskName, false)
-    } yield ()
+  def run(ctx: Context) =
+    migratePeriodicItemTasks.transact(mkTransactor(ctx))
 
   def migratePeriodicItemTasks: ConnectionIO[Unit] =
     for {
@@ -137,75 +113,4 @@ trait MigrationTasks {
       ref = r.asRef
     } yield ref)
       .getOrElseF(Sync[ConnectionIO].raiseError(new Exception("User not found!")))
-
-  private def migrateDueItemTask1(old: RPeriodicTask): ConnectionIO[Int] = {
-    val converted = old.args
-      .parseJsonAs[NotifyDueItemsArgs]
-      .leftMap(_.getMessage())
-      .map(convertArgs)
-
-    converted match {
-      case Right(args) =>
-        val task = args
-          .semiflatMap(a =>
-            RPeriodicTask
-              .updateTask(
-                old.id,
-                PeriodicDueItemsArgsLegacy.taskName,
-                a.asJson.noSpaces
-              )
-          )
-          .getOrElse(0)
-
-        Sync[ConnectionIO].delay(logger.info(s"Converting user task: $old")) *> task
-
-      case Left(err) =>
-        logger.error(s"Error converting user task: $old. $err")
-        0.pure[ConnectionIO]
-    }
-  }
-
-  private def convertArgs(
-      old: NotifyDueItemsArgs
-  ): OptionT[ConnectionIO, PeriodicDueItemsArgsLegacy] = {
-    val recs = old.recipients
-      .map(MailAddress.parse)
-      .flatMap {
-        case Right(m) => Some(m)
-        case Left(err) =>
-          logger.warn(s"Cannot read mail address: $err. Skip this while migrating.")
-          None
-      }
-
-    for {
-      userData <- OptionT(QLogin.findAccount(old.account))
-      userId = userData.userId
-      id <- OptionT.liftF(Ident.randomId[ConnectionIO])
-      now <- OptionT.liftF(Timestamp.current[ConnectionIO])
-      chName = Some("migrate notify items")
-      ch = RNotificationChannelMail(
-        id,
-        userId,
-        chName,
-        old.smtpConnection,
-        recs,
-        now
-      )
-      _ <- OptionT.liftF(RNotificationChannelMail.insert(ch))
-      args = PeriodicDueItemsArgsLegacy(
-        old.account,
-        NonEmptyList.of(ChannelRef(ch.id, ChannelType.Mail, chName)),
-        old.remindDays,
-        old.daysBack,
-        old.tagsInclude,
-        old.tagsExclude,
-        old.itemDetailUrl
-      )
-    } yield args
-  }
-
-  def mkTransactor(ctx: Context): Transactor[IO] = {
-    val xa = Transactor.fromConnection[IO](ctx.getConnection)
-    Transactor.strategy.set(xa, Strategy.void) // transactions are handled by flyway
-  }
 }

modules/store/src/main/scala/db/migration/common/MigrateNotifyTasks.scala

@@ -0,0 +1,102 @@
+package db.migration.common
+
+import cats.data.{NonEmptyList, OptionT}
+import cats.effect.{IO, Sync}
+import cats.implicits._
+
+import docspell.common._
+import docspell.common.syntax.StringSyntax._
+import docspell.notification.api._
+import docspell.store.queries.QLogin
+import docspell.store.records._
+
+import db.migration.data.{PeriodicDueItemsArgs => PeriodicDueItemsArgsLegacy, _}
+import doobie._
+import doobie.implicits._
+import emil.MailAddress
+import emil.javamail.syntax._
+import io.circe.syntax._
+import org.flywaydb.core.api.migration.Context
+
+object MigrateNotifyTasks extends TransactorSupport {
+  val logger = docspell.logging.getLogger[IO]
+
+  def run(ctx: Context): IO[Unit] =
+    migrateDueItemTasks.transact(mkTransactor(ctx))
+
+  def migrateDueItemTasks: ConnectionIO[Unit] =
+    for {
+      tasks <- RPeriodicTask.findByTask(NotifyDueItemsArgs.taskName)
+      _ <- Sync[ConnectionIO].delay(
+        logger.info(s"Starting to migrate ${tasks.size} user tasks")
+      )
+      _ <- tasks.traverse(migrateDueItemTask1)
+      _ <- RPeriodicTask.setEnabledByTask(NotifyDueItemsArgs.taskName, false)
+    } yield ()
+
+  private def migrateDueItemTask1(old: RPeriodicTask): ConnectionIO[Int] = {
+    val converted = old.args
+      .parseJsonAs[NotifyDueItemsArgs]
+      .leftMap(_.getMessage())
+      .map(convertArgs)
+
+    converted match {
+      case Right(args) =>
+        val task = args
+          .semiflatMap(a =>
+            RPeriodicTask
+              .updateTask(
+                old.id,
+                PeriodicDueItemsArgsLegacy.taskName,
+                a.asJson.noSpaces
+              )
+          )
+          .getOrElse(0)
+
+        Sync[ConnectionIO].delay(logger.info(s"Converting user task: $old")) *> task
+
+      case Left(err) =>
+        logger.error(s"Error converting user task: $old. $err")
+        0.pure[ConnectionIO]
+    }
+  }
+
+  private def convertArgs(
+      old: NotifyDueItemsArgs
+  ): OptionT[ConnectionIO, PeriodicDueItemsArgsLegacy] = {
+    val recs = old.recipients
+      .map(MailAddress.parse)
+      .flatMap {
+        case Right(m) => Some(m)
+        case Left(err) =>
+          logger.warn(s"Cannot read mail address: $err. Skip this while migrating.")
+          None
+      }
+
+    for {
+      userData <- OptionT(QLogin.findAccount(old.account))
+      userId = userData.userId
+      id <- OptionT.liftF(Ident.randomId[ConnectionIO])
+      now <- OptionT.liftF(Timestamp.current[ConnectionIO])
+      chName = Some("migrate notify items")
+      ch = RNotificationChannelMail(
+        id,
+        userId,
+        chName,
+        old.smtpConnection,
+        recs,
+        now
+      )
+      _ <- OptionT.liftF(RNotificationChannelMail.insert(ch))
+      args = PeriodicDueItemsArgsLegacy(
+        old.account,
+        NonEmptyList.of(ChannelRef(ch.id, ChannelType.Mail, chName)),
+        old.remindDays,
+        old.daysBack,
+        old.tagsInclude,
+        old.tagsExclude,
+        old.itemDetailUrl
+      )
+    } yield args
+  }
+}

modules/store/src/main/scala/db/migration/common/TransactorSupport.scala

@@ -0,0 +1,17 @@
+package db.migration.common
+
+import cats.effect.IO
+import docspell.logging.Logger
+import doobie.util.transactor.{Strategy, Transactor}
+import org.flywaydb.core.api.migration.Context
+
+trait TransactorSupport {
+
+  def logger: Logger[IO]
+
+  def mkTransactor(ctx: Context): Transactor[IO] = {
+    val xa = Transactor.fromConnection[IO](ctx.getConnection)
+    logger.asUnsafe.info(s"Creating transactor for db migrations from connection: $xa")
+    Transactor.strategy.set(xa, Strategy.void) // transactions are handled by flyway
+  }
+}

modules/store/src/main/scala/db/migration/h2/V1_29_2__MigrateNotifyTask.scala

@@ -7,17 +7,13 @@
 package db.migration.h2
 
 import cats.effect.unsafe.implicits._
-
-import db.migration.MigrationTasks
-import doobie.implicits._
+import db.migration.common.MigrateNotifyTasks
 import org.flywaydb.core.api.migration.BaseJavaMigration
 import org.flywaydb.core.api.migration.Context
 
-class V1_29_2__MigrateNotifyTask extends BaseJavaMigration with MigrationTasks {
+class V1_29_2__MigrateNotifyTask extends BaseJavaMigration {
   val logger = org.log4s.getLogger
 
-  override def migrate(ctx: Context): Unit = {
-    val xa = mkTransactor(ctx)
-    migrateDueItemTasks.transact(xa).unsafeRunSync()
-  }
+  override def migrate(ctx: Context): Unit =
+    MigrateNotifyTasks.run(ctx).unsafeRunSync()
 }

modules/store/src/main/scala/db/migration/h2/V1_32_2__MigrateChannels.scala

@@ -7,16 +7,12 @@
 package db.migration.h2
 
 import cats.effect.unsafe.implicits._
-
-import db.migration.MigrationTasks
-import doobie.implicits._
+import db.migration.common.MigrateDueItemTasks
 import org.flywaydb.core.api.migration.{BaseJavaMigration, Context}
 
-class V1_32_2__MigrateChannels extends BaseJavaMigration with MigrationTasks {
+class V1_32_2__MigrateChannels extends BaseJavaMigration {
   val logger = org.log4s.getLogger
 
-  override def migrate(ctx: Context): Unit = {
-    val xa = mkTransactor(ctx)
-    migratePeriodicItemTasks.transact(xa).unsafeRunSync()
-  }
+  override def migrate(ctx: Context): Unit =
+    MigrateDueItemTasks.run(ctx).unsafeRunSync()
 }

modules/store/src/main/scala/db/migration/mariadb/V1_29_2__MigrateNotifyTask.scala

@@ -7,17 +7,13 @@
 package db.migration.mariadb
 
 import cats.effect.unsafe.implicits._
-
-import db.migration.MigrationTasks
-import doobie.implicits._
+import db.migration.common.MigrateNotifyTasks
 import org.flywaydb.core.api.migration.BaseJavaMigration
 import org.flywaydb.core.api.migration.Context
 
-class V1_29_2__MigrateNotifyTask extends BaseJavaMigration with MigrationTasks {
+class V1_29_2__MigrateNotifyTask extends BaseJavaMigration {
   val logger = org.log4s.getLogger
 
-  override def migrate(ctx: Context): Unit = {
-    val xa = mkTransactor(ctx)
-    migrateDueItemTasks.transact(xa).unsafeRunSync()
-  }
+  override def migrate(ctx: Context): Unit =
+    MigrateNotifyTasks.run(ctx).unsafeRunSync()
 }

modules/store/src/main/scala/db/migration/mariadb/V1_32_2__MigrateChannels.scala

@@ -7,16 +7,13 @@
 package db.migration.mariadb
 
 import cats.effect.unsafe.implicits._
-
-import db.migration.MigrationTasks
-import doobie.implicits._
+import db.migration.common.MigrateDueItemTasks
 import org.flywaydb.core.api.migration.{BaseJavaMigration, Context}
 
-class V1_32_2__MigrateChannels extends BaseJavaMigration with MigrationTasks {
+class V1_32_2__MigrateChannels extends BaseJavaMigration {
   val logger = org.log4s.getLogger
 
-  override def migrate(ctx: Context): Unit = {
-    val xa = mkTransactor(ctx)
-    migratePeriodicItemTasks.transact(xa).unsafeRunSync()
-  }
+  override def migrate(ctx: Context): Unit =
+    MigrateDueItemTasks.run(ctx).unsafeRunSync()
+
 }

modules/store/src/main/scala/db/migration/postgresql/V1_29_2__MigrateNotifyTask.scala

@@ -7,17 +7,13 @@
 package db.migration.postgresql
 
 import cats.effect.unsafe.implicits._
-
-import db.migration.MigrationTasks
-import doobie.implicits._
+import db.migration.common.MigrateNotifyTasks
 import org.flywaydb.core.api.migration.BaseJavaMigration
 import org.flywaydb.core.api.migration.Context
 
-class V1_29_2__MigrateNotifyTask extends BaseJavaMigration with MigrationTasks {
+class V1_29_2__MigrateNotifyTask extends BaseJavaMigration {
   val logger = org.log4s.getLogger
 
-  override def migrate(ctx: Context): Unit = {
-    val xa = mkTransactor(ctx)
-    migrateDueItemTasks.transact(xa).unsafeRunSync()
-  }
+  override def migrate(ctx: Context): Unit =
+    MigrateNotifyTasks.run(ctx).unsafeRunSync()
 }

modules/store/src/main/scala/db/migration/postgresql/V1_32_2__MigrateChannels.scala

@@ -7,16 +7,12 @@
 package db.migration.postgresql
 
 import cats.effect.unsafe.implicits._
-
-import db.migration.MigrationTasks
-import doobie.implicits._
+import db.migration.common.MigrateDueItemTasks
 import org.flywaydb.core.api.migration.{BaseJavaMigration, Context}
 
-class V1_32_2__MigrateChannels extends BaseJavaMigration with MigrationTasks {
+class V1_32_2__MigrateChannels extends BaseJavaMigration {
   val logger = org.log4s.getLogger
 
-  override def migrate(ctx: Context): Unit = {
-    val xa = mkTransactor(ctx)
-    migratePeriodicItemTasks.transact(xa).unsafeRunSync()
-  }
+  override def migrate(ctx: Context): Unit =
+    MigrateDueItemTasks.run(ctx).unsafeRunSync()
 }