Merge pull request #1690 from eikek/h2-update

Update h2 and flyway

docker/dockerfiles/joex.dockerfile

@@ -53,7 +53,7 @@ RUN JDKPKG="openjdk11-jre"; \
     qpdf-dev \
     openssl-dev \
     ocrmypdf \
-  && apk add 'zlib=1.2.12-r2' \
+  && apk add 'zlib=1.2.12-r3' \
   && pip3 install --upgrade pip \
   && pip3 install ocrmypdf \
   && curl -Ls $UNO_URL -o /usr/local/bin/unoconv \

docker/dockerfiles/restserver.dockerfile

@@ -8,7 +8,7 @@ RUN JDKPKG="openjdk11-jre"; \
     if [[ $TARGETPLATFORM = linux/arm* ]]; then JDKPKG="openjdk8-jre"; fi; \
     apk update && \
     apk add --no-cache $JDKPKG bash tzdata && \
-    apk add 'zlib=1.2.12-r2'
+    apk add 'zlib=1.2.12-r3'
 
 WORKDIR /opt
 RUN wget ${restserver_url:-https://github.com/eikek/docspell/releases/download/v$version/docspell-restserver-$version.zip} && \

modules/scheduler/impl/src/main/scala/docspell/scheduler/impl/QJob.scala

@@ -89,7 +89,7 @@ object QJob {
       res <- job.traverse(j => markJob(j))
     } yield res.map(_.map(_.some)).getOrElse {
       if (group.isDefined)
-        Left(()) // if a group was found, but no job someone else was faster
+        Left(()) // if a group was found but no job, someone else was faster
       else Right(None)
     }
   }
@@ -115,33 +115,27 @@ object QJob {
       val selectAll = Select(JC.group.s, from(JC), stateCond).distinct
     }
 
-    val sql1 =
-      Select(
-        select(min(AllGroups.group).as("g"), lit("0 as n")),
-        from(AllGroups),
-        AllGroups.group > Select(G.group.s, from(G), G.worker === worker)
-      )
-
-    val sql2 =
-      Select(
-        select(min(AllGroups.group).as("g"), lit("1 as n")),
-        from(AllGroups)
-      )
-
-    val gcol = Column[String]("g", TableDef(""))
-    val gnum = Column[Int]("n", TableDef(""))
     val groups =
       withCte(AllGroups -> AllGroups.selectAll)
-        .select(Select(gcol.s, from(union(sql1, sql2), "t0"), gcol.isNull.negate))
-        .orderBy(gnum.asc)
-        .limit(1)
+        .select(
+          Select(
+            coalesce(
+              Select(
+                select(min(AllGroups.group)),
+                from(AllGroups),
+                AllGroups.group > Select(G.group.s, from(G), G.worker === worker)
+              ).asSubSelect,
+              Select(select(min(AllGroups.group)), from(AllGroups)).asSubSelect
+            ).s
+          )
+        )
 
     val frag = groups.build
-    cioLogger.trace(
-      s"nextGroupQuery: $frag  (now=${now.toMillis}, pause=${initialPause.millis})"
-    )
-
-    frag.query[Ident].option
+    cioLogger
+      .trace(
+        s"nextGroupQuery: $frag  (now=${now.toMillis}, pause=${initialPause.millis})"
+      ) *>
+      groups.build.query[Ident].option
   }
 
   private def stuckTriggerValue(t: RJob.Table, initialPause: Duration, now: Timestamp) =

modules/scheduler/impl/src/test/scala/docspell/scheduler/impl/QJobTest.scala

@@ -9,17 +9,13 @@ package docspell.scheduler.impl
 import java.time.Instant
 import java.util.concurrent.atomic.AtomicLong
 
-import cats.implicits._
+import cats.syntax.all._
 
 import docspell.common._
-import docspell.logging.TestLoggingConfig
-import docspell.store.StoreFixture
 import docspell.store.records.{RJob, RJobGroupUse}
+import docspell.store.{DatabaseTest, Db}
 
-import doobie.implicits._
-import munit._
-
-class QJobTest extends CatsEffectSuite with StoreFixture with TestLoggingConfig {
+class QJobTest extends DatabaseTest {
   private[this] val c = new AtomicLong(0)
 
   private val worker = Ident.unsafe("joex1")
@@ -28,6 +24,11 @@ class QJobTest extends CatsEffectSuite with StoreFixture with TestLoggingConfig
   private val group1 = Ident.unsafe("group1")
   private val group2 = Ident.unsafe("group2")
 
+  override def munitFixtures = h2File ++ mariaDbAll ++ postgresAll
+
+  def createStore(dbms: Db) =
+    dbms.fold(pgStore(), mariaStore(), h2FileStore())
+
   def createJob(group: Ident): RJob =
     RJob.fromJson[Unit](
       Ident.unsafe(s"job-${c.incrementAndGet()}"),
@@ -41,54 +42,66 @@ class QJobTest extends CatsEffectSuite with StoreFixture with TestLoggingConfig
       None
     )
 
-  xa.test("set group must insert or update") { tx =>
-    val res =
-      for {
-        _ <- RJobGroupUse.setGroup(RJobGroupUse(group1, worker)).transact(tx)
-        res <- RJobGroupUse.findGroup(worker).transact(tx)
-      } yield res
-
-    res.assertEquals(Some(group1))
+  Db.all.toList.foreach { db =>
+    test(s"set group must insert or update ($db)") {
+      val store = createStore(db)
+      val res =
+        for {
+          _ <- store.transact(RJobGroupUse.setGroup(RJobGroupUse(group1, worker)))
+          res <- store.transact(RJobGroupUse.findGroup(worker))
+        } yield res
+
+      res.assertEquals(Some(group1))
+    }
   }
 
-  xa.test("selectNextGroup should return first group on initial state") { tx =>
-    val nextGroup = for {
-      _ <- List(group1, group2, group1, group2, group2)
-        .map(createJob)
-        .map(RJob.insert)
-        .traverse(_.transact(tx))
-      _ <- RJobGroupUse.deleteAll.transact(tx)
-      next <- QJob.selectNextGroup(worker, nowTs, initialPause).transact(tx)
-    } yield next
-
-    nextGroup.assertEquals(Some(group1))
+  Db.all.toList.foreach { db =>
+    test(s"selectNextGroup should return first group on initial state ($db)") {
+      val store = createStore(db)
+      val nextGroup = for {
+        _ <- List(group1, group2, group1, group2, group2)
+          .map(createJob)
+          .map(RJob.insert)
+          .traverse_(store.transact(_))
+        _ <- store.transact(RJobGroupUse.deleteAll)
+        next <- store.transact(QJob.selectNextGroup(worker, nowTs, initialPause))
+      } yield next
+
+      nextGroup.assertEquals(Some(group1))
+    }
   }
 
-  xa.test("selectNextGroup should return second group on subsequent call (1)") { tx =>
-    val nextGroup = for {
-      _ <- List(group1, group2, group1, group2)
-        .map(createJob)
-        .map(RJob.insert)
-        .traverse(_.transact(tx))
-      _ <- RJobGroupUse.deleteAll.transact(tx)
-      _ <- RJobGroupUse.setGroup(RJobGroupUse(group1, worker)).transact(tx)
-      next <- QJob.selectNextGroup(worker, nowTs, initialPause).transact(tx)
-    } yield next
-
-    nextGroup.assertEquals(Some(group2))
+  Db.all.toList.foreach { db =>
+    test(s"selectNextGroup should return second group on subsequent call ($db)") {
+      val store = createStore(db)
+      val nextGroup = for {
+        _ <- List(group1, group2, group1, group2)
+          .map(createJob)
+          .map(RJob.insert)
+          .traverse_(store.transact(_))
+        _ <- store.transact(RJobGroupUse.deleteAll)
+        _ <- store.transact(RJobGroupUse.setGroup(RJobGroupUse(group1, worker)))
+        next <- store.transact(QJob.selectNextGroup(worker, nowTs, initialPause))
+      } yield next
+
+      nextGroup.assertEquals(Some(group2))
+    }
   }
 
-  xa.test("selectNextGroup should return second group on subsequent call (2)") { tx =>
-    val nextGroup = for {
-      _ <- List(group1, group2, group1, group2)
-        .map(createJob)
-        .map(RJob.insert)
-        .traverse(_.transact(tx))
-      _ <- RJobGroupUse.deleteAll.transact(tx)
-      _ <- RJobGroupUse.setGroup(RJobGroupUse(group2, worker)).transact(tx)
-      next <- QJob.selectNextGroup(worker, nowTs, initialPause).transact(tx)
-    } yield next
-
-    nextGroup.assertEquals(Some(group1))
+  Db.all.toList.foreach { db =>
+    test(s"selectNextGroup should return first group on subsequent call ($db)") {
+      val store = createStore(db)
+      val nextGroup = for {
+        _ <- List(group1, group2, group1, group2)
+          .map(createJob)
+          .map(RJob.insert)
+          .traverse_(store.transact(_))
+        _ <- store.transact(RJobGroupUse.deleteAll)
+        _ <- store.transact(RJobGroupUse.setGroup(RJobGroupUse(group2, worker)))
+        next <- store.transact(QJob.selectNextGroup(worker, nowTs, initialPause))
+      } yield next
+
+      nextGroup.assertEquals(Some(group1))
+    }
   }
 }

modules/store/src/main/resources/db/migration/h2/V1.39.0__collective_id.sql

@@ -1,5 +1,5 @@
 -- add new id column
-alter table "collective" add column "id" bigserial not null;
+alter table "collective" add column "id" bigserial not null unique;
 create unique index "collective_id_idx" on "collective"("id");
 
 -- change references: source

modules/store/src/main/resources/db/migration/h2/V1.39.3__rename_value.sql

@@ -0,0 +1 @@
+alter table "contact" rename column "value" to "value_";

modules/store/src/main/resources/db/migration/h2/V1.9.3__joblog_counter.sql

@@ -1,2 +1,2 @@
 ALTER TABLE "joblog"
-ADD COLUMN "counter" bigint auto_increment;
+ADD COLUMN "counter" bigint generated always as identity;

modules/store/src/main/resources/db/migration/mariadb/V1.39.3__rename_value.sql

@@ -0,0 +1 @@
+alter table `contact` rename column `value` to `value_`;

modules/store/src/main/resources/db/migration/postgresql/V1.39.3__rename_value.sql

@@ -0,0 +1 @@
+alter table "contact" rename column "value" to "value_";

modules/store/src/main/scala/docspell/store/qb/Select.scala

@@ -7,6 +7,7 @@
 package docspell.store.qb
 
 import cats.data.{NonEmptyList => Nel}
+import cats.syntax.option._
 
 import docspell.store.qb.impl.SelectBuilder
 
@@ -23,6 +24,9 @@ sealed trait Select {
   def as(alias: String): SelectExpr.SelectQuery =
     SelectExpr.SelectQuery(this, Some(alias))
 
+  def asSubSelect: SelectExpr.SelectQuery =
+    SelectExpr.SelectQuery(this, None)
+
   /** Adds one or more order-by definitions */
   def orderBy(ob: OrderBy, obs: OrderBy*): Select
 
@@ -71,35 +75,38 @@ sealed trait Select {
 }
 
 object Select {
+  def apply(projection: SelectExpr) =
+    SimpleSelect(false, Nel.of(projection), None, Condition.unit, None)
+
   def apply(projection: Nel[SelectExpr], from: FromExpr) =
-    SimpleSelect(false, projection, from, Condition.unit, None)
+    SimpleSelect(false, projection, from.some, Condition.unit, None)
 
   def apply(projection: SelectExpr, from: FromExpr) =
-    SimpleSelect(false, Nel.of(projection), from, Condition.unit, None)
+    SimpleSelect(false, Nel.of(projection), from.some, Condition.unit, None)
 
   def apply(
       projection: Nel[SelectExpr],
       from: FromExpr,
       where: Condition
-  ) = SimpleSelect(false, projection, from, where, None)
+  ) = SimpleSelect(false, projection, from.some, where, None)
 
   def apply(
       projection: SelectExpr,
       from: FromExpr,
       where: Condition
-  ) = SimpleSelect(false, Nel.of(projection), from, where, None)
+  ) = SimpleSelect(false, Nel.of(projection), from.some, where, None)
 
   def apply(
       projection: Nel[SelectExpr],
       from: FromExpr,
       where: Condition,
       groupBy: GroupBy
-  ) = SimpleSelect(false, projection, from, where, Some(groupBy))
+  ) = SimpleSelect(false, projection, from.some, where, Some(groupBy))
 
   case class SimpleSelect(
       distinctFlag: Boolean,
       projection: Nel[SelectExpr],
-      from: FromExpr,
+      from: Option[FromExpr],
       where: Condition,
       groupBy: Option[GroupBy]
   ) extends Select {
@@ -125,7 +132,7 @@ object Select {
       copy(projection = es)
 
     def changeFrom(f: FromExpr => FromExpr): SimpleSelect =
-      copy(from = f(from))
+      copy(from = from.map(f))
 
     def changeWhere(f: Condition => Condition): SimpleSelect =
       copy(where = f(where))

modules/store/src/main/scala/docspell/store/qb/SelectExpr.scala

@@ -33,6 +33,8 @@ object SelectExpr {
   case class SelectLiteral(value: String, alias: Option[String]) extends SelectExpr {
     def as(a: String): SelectLiteral =
       copy(alias = Some(a))
+    def as(otherCol: Column[_]): SelectExpr =
+      copy(alias = Some(otherCol.name))
   }
 
   case class SelectQuery(query: Select, alias: Option[String]) extends SelectExpr {

modules/store/src/main/scala/docspell/store/qb/impl/SelectBuilder.scala

@@ -52,7 +52,7 @@ object SelectBuilder {
 
   def buildSimple(sq: Select.SimpleSelect): Fragment = {
     val f0 = sq.projection.map(selectExpr).reduceLeft(_ ++ comma ++ _)
-    val f1 = fromExpr(sq.from)
+    val f1 = sq.from.map(fromExpr).getOrElse(Fragment.empty)
     val f2 = cond(sq.where)
     val f3 = sq.groupBy.map(groupBy).getOrElse(Fragment.empty)
     f0 ++ f1 ++ f2 ++ f3

modules/store/src/main/scala/docspell/store/records/RContact.scala

@@ -30,7 +30,7 @@ object RContact {
     val tableName = "contact"
 
     val contactId = Column[Ident]("contactid", this)
-    val value = Column[String]("value", this)
+    val value = Column[String]("value_", this)
     val kind = Column[ContactKind]("kind", this)
     val personId = Column[Ident]("pid", this)
     val orgId = Column[Ident]("oid", this)

modules/store/src/main/scala/docspell/store/records/RJob.scala

@@ -300,11 +300,6 @@ object RJob {
   def setProgress(jobId: Ident, perc: Int): ConnectionIO[Int] =
     DML.update(T, T.id === jobId, DML.set(T.progress.setTo(perc)))
 
-  def selectWaiting: ConnectionIO[Option[RJob]] = {
-    val sql = run(select(T.all), from(T), T.state === JobState.waiting)
-    sql.query[RJob].to[Vector].map(_.headOption)
-  }
-
   def selectGroupInState(states: NonEmptyList[JobState]): ConnectionIO[Vector[Ident]] = {
     val sql =
       Select(select(T.group), from(T), T.state.in(states)).orderBy(T.group)