[JSON-API] Rework prior work and introduce the object SchemaHandling

changelog_begin - [JSON-API] Schema versioning was introduced to the db schema. Because of this the field `createSchema` in the jdbcConfig was removed and replaced by the field `schemaHandling` which can have the following values: 1. `ForceCreateAndTerminate`: This is equal to the behaviour of `createSchema=true` so the db schema is created and then the application terminates. 2. `CheckAndTerminateIfWrong`: With this the schema version is checked and if no version or an outdated version was found the application terminates. 3. `CreateOrUpdateAndContinue`: With this the schema version is checked and if no version or an outdated version was found the schema will be created/updated and the application proceeds with the startup. 4. `ForceCreateAndContinue`: Similar to the first option but instead of terminating the application proceeds with the startup. changelog_end
digital-asset · Jul 23, 2021 · f4c6b7c · f4c6b7c
1 parent 9938b6a
commit f4c6b7c
Show file tree

Hide file tree

Showing 12 changed files with 139 additions and 59 deletions.
diff --git a/ledger-service/http-json-cli/src/main/scala/com/daml/http/Config.scala b/ledger-service/http-json-cli/src/main/scala/com/daml/http/Config.scala
@@ -113,16 +113,14 @@ private[http] final case class JdbcConfig(
     url: String,
     user: String,
     password: String,
-    createSchema: Boolean = false,
-    checkIfExists: Boolean = false,
-    continueAfterSchemaCreation: Boolean = false,
+    schemaHandling: SchemaHandling = SchemaHandling.CheckAndTerminateIfWrong,
 )
 
 private[http] object JdbcConfig
     extends ConfigCompanion[JdbcConfig, Config.SupportedJdbcDriverNames]("JdbcConfig") {
 
   implicit val showInstance: Show[JdbcConfig] = Show.shows(a =>
-    s"JdbcConfig(driver=${a.driver}, url=${a.url}, user=${a.user}, createSchema=${a.createSchema})"
+    s"JdbcConfig(driver=${a.driver}, url=${a.url}, user=${a.user}, schemaHandling=${a.schemaHandling})"
   )
 
   def help(implicit supportedJdbcDriverNames: Config.SupportedJdbcDriverNames): String =
@@ -131,27 +129,21 @@ private[http] object JdbcConfig
       s"${indent}url -- JDBC connection URL,\n" +
       s"${indent}user -- database user name,\n" +
       s"${indent}password -- database user password,\n" +
-      s"${indent}createSchema -- boolean flag, if set to true, the process will re-create database schema and terminate immediately if the continueAfterSchemaCreation flag was not set.\n" +
-      s"${indent}checkIfExists -- boolean flag, if set to true, the process will check during re-creating the database schema if the schema already exists & leave it untouched.\n" +
-      s"${indent}continueAfterSchemaCreation -- boolean flag, if set to true, the process won't terminate after the database schema was created.\n" +
+      s"${indent}schemaHandling -- option setting how the schema should be handled. Valid options are ForceCreateAndTerminate, CheckAndTerminateIfWrong, CreateOrUpdateAndContinue, ForceCreateAndContinue.\n" +
       s"${indent}Example: " + helpString(
         "org.postgresql.Driver",
         "jdbc:postgresql://localhost:5432/test?&ssl=true",
         "postgres",
         "password",
-        "false",
-        "false",
-        "false",
+        "ForceCreateAndTerminate",
       )
 
   lazy val usage: String = helpString(
     "<JDBC driver class name>",
     "<JDBC connection url>",
     "<user>",
     "<password>",
-    "<true|false>",
-    "<true|false>",
-    "<true|false>",
+    "<ForceCreateAndTerminate|CheckAndTerminateIfWrong|CreateOrUpdateAndContinue|ForceCreateAndContinue>",
   )
 
   override def create(x: Map[String, String])(implicit
@@ -168,29 +160,23 @@ private[http] object JdbcConfig
       url <- requiredField(x)("url")
       user <- requiredField(x)("user")
       password <- requiredField(x)("password")
-      createSchema <- optionalBooleanField(x)("createSchema")
-      checkIfExists <- optionalBooleanField(x)("checkIfExists")
-      continueAfterSchemaCreation <- optionalBooleanField(x)("continueAfterSchemaCreation")
+      schemaHandling <- SchemaHandling.optionalSchemaHandlingField(x)("schemaHandling")
     } yield JdbcConfig(
       driver = driver,
       url = url,
       user = user,
       password = password,
-      createSchema = createSchema.getOrElse(false),
-      checkIfExists = checkIfExists.getOrElse(false),
-      continueAfterSchemaCreation = continueAfterSchemaCreation.getOrElse(false),
+      schemaHandling = schemaHandling.getOrElse(SchemaHandling.CheckAndTerminateIfWrong),
     )
 
   private def helpString(
       driver: String,
       url: String,
       user: String,
       password: String,
-      createSchema: String,
-      checkIfExists: String,
-      continueAfterSchemaCreation: String,
+      schemaHandling: String,
   ): String =
-    s"""\"driver=$driver,url=$url,user=$user,password=$password,createSchema=$createSchema,checkIfExists=$checkIfExists,continueAfterSchemaCreation=$continueAfterSchemaCreation\""""
+    s"""\"driver=$driver,url=$url,user=$user,password=$password,schemaHandling=$schemaHandling\""""
 }
 
 // It is public for Daml Hub

diff --git a/ledger-service/http-json-cli/src/main/scala/com/daml/http/SchemaHandling.scala b/ledger-service/http-json-cli/src/main/scala/com/daml/http/SchemaHandling.scala
@@ -0,0 +1,27 @@
+package com.daml.http
+import scalaz.std.option._
+import scalaz.syntax.traverse._
+
+private[http] sealed trait SchemaHandling
+private[http] object SchemaHandling {
+  private[http] case object ForceCreateAndTerminate extends SchemaHandling
+  private[http] case object CheckAndTerminateIfWrong extends SchemaHandling
+  private[http] case object CreateOrUpdateAndContinue extends SchemaHandling
+  private[http] case object ForceCreateAndContinue extends SchemaHandling
+
+  import scalaz.Validation.{success, failure}
+  import scalaz.Validation
+
+  private[http] def optionalSchemaHandlingField(
+      m: Map[String, String]
+  )(k: String): Either[String, Option[SchemaHandling]] = {
+    val parse: String => Validation[String, SchemaHandling] = {
+      case "ForceCreateAndTerminate" => success(ForceCreateAndTerminate)
+      case "CheckAndTerminateIfWrong" => success(CheckAndTerminateIfWrong)
+      case "CreateOrUpdateAndContinue" => success(CreateOrUpdateAndContinue)
+      case "ForceCreateAndContinue" => success(ForceCreateAndContinue)
+      case opt => failure(s"Unrecognized option $opt")
+    }
+    m.get(k).traverse(input => parse(input).disjunction).toEither
+  }
+}
diff --git a/ledger-service/http-json-perf/src/main/scala/com/daml/http/perf/Main.scala b/ledger-service/http-json-perf/src/main/scala/com/daml/http/perf/Main.scala
@@ -224,7 +224,6 @@ object Main extends StrictLogging {
       url = c.url,
       user = "test",
       password = "",
-      createSchema = true,
     )
 
   private def resolveSimulationClass(str: String): Throwable \/ Class[_ <: Simulation] = {

diff --git a/ledger-service/http-json-testing/src/main/scala/com/daml/http/HttpServiceTestFixture.scala b/ledger-service/http-json-testing/src/main/scala/com/daml/http/HttpServiceTestFixture.scala
@@ -240,7 +240,7 @@ object HttpServiceTestFixture extends LazyLogging with Assertions with Inside {
       _ <- {
         import dao.{jdbcDriver, logHandler}
         dao
-          .transact(ContractDao.initialize(checkIfExists = c.checkIfExists))
+          .transact(ContractDao.initialize)
           .unsafeToFuture(): Future[Unit]
       }
     } yield dao

diff --git a/ledger-service/http-json/src/bench/scala/com/daml/http/dbbackend/ContractDaoBenchmark.scala b/ledger-service/http-json/src/bench/scala/com/daml/http/dbbackend/ContractDaoBenchmark.scala
@@ -37,7 +37,7 @@ abstract class ContractDaoBenchmark extends OracleAround {
 
     import oracleDao.jdbcDriver
 
-    dao.transact(ContractDao.initialize(checkIfExists = false)).unsafeRunSync()
+    dao.transact(ContractDao.initialize).unsafeRunSync()
   }
 
   @TearDown(Level.Trial)

diff --git a/ledger-service/http-json/src/failure/scala/http/HttpTestFixture.scala b/ledger-service/http-json/src/failure/scala/http/HttpTestFixture.scala
@@ -38,7 +38,6 @@ trait HttpFailureTestFixture extends ToxicSandboxFixture with PostgresAroundAll
         s"jdbc:postgresql://${postgresDatabase.hostName}:$dbProxyPort/${postgresDatabase.databaseName}?user=${postgresDatabase.userName}&password=${postgresDatabase.password}",
       user = "test",
       password = "",
-      createSchema = true,
     )
 
   override def packageFiles =

diff --git a/ledger-service/http-json/src/it/scala/http/HttpServicePostgresInt.scala b/ledger-service/http-json/src/it/scala/http/HttpServicePostgresInt.scala
@@ -19,6 +19,5 @@ trait HttpServicePostgresInt extends AbstractHttpServiceIntegrationTestFuns with
     url = postgresDatabase.url,
     user = "test",
     password = "",
-    createSchema = true,
   )
 }
diff --git a/ledger-service/http-json/src/it/scala/http/HttpServiceWithPostgresIntTest.scala b/ledger-service/http-json/src/it/scala/http/HttpServiceWithPostgresIntTest.scala
@@ -3,7 +3,10 @@
 
 package com.daml.http
 
-import com.daml.http.dbbackend.{ContractDao, SupportedJdbcDriver}
+import com.daml.http.SchemaHandlingResult.Continue
+import com.daml.http.dbbackend.SupportedJdbcDriver
+import com.daml.http.util.Logging.{InstanceUUID, instanceUUIDLogCtx}
+import com.daml.logging.LoggingContextOf
 import com.daml.scalautil.Statement.discard
 import com.daml.testing.postgresql.PostgresAroundAll
 import doobie.util.log
@@ -34,12 +37,21 @@ class HttpServiceWithPostgresIntTest
 
   "if the schema version in the db is not equal to the current one the cache will be re-initialized" in {
     import doobie.implicits.toSqlInterpolator
+    implicit val lc: LoggingContextOf[InstanceUUID] = instanceUUIDLogCtx()
     val res =
       for {
-        _ <- dao.transact(ContractDao.initialize(checkIfExists = false))
+        res1 <- SchemaHandlingResult.fromSchemaHandling(
+          dao,
+          SchemaHandling.ForceCreateAndContinue,
+        )
+        _ = res1 shouldBe Continue
         _ <- dao.transact(sql"DELETE FROM json_api_schema_version".update.run)
         _ <- dao.transact(sql"INSERT INTO json_api_schema_version(version) VALUES(0)".update.run)
-        _ <- dao.transact(ContractDao.initialize(checkIfExists = true))
+        res2 <- SchemaHandlingResult.fromSchemaHandling(
+          dao,
+          SchemaHandling.CreateOrUpdateAndContinue,
+        )
+        _ = res2 shouldBe Continue
         version <- dao.transact(sql"SELECT version FROM json_api_schema_version".query[Int].unique)
       } yield {
         version should not be 0
@@ -50,11 +62,20 @@ class HttpServiceWithPostgresIntTest
 
   "if the schema version in the db is equal to the current one the cache won't be re-initialized" in {
     import doobie.implicits.toSqlInterpolator
+    implicit val lc: LoggingContextOf[InstanceUUID] = instanceUUIDLogCtx()
     val res =
       for {
-        _ <- dao.transact(ContractDao.initialize(checkIfExists = false))
+        res1 <- SchemaHandlingResult.fromSchemaHandling(
+          dao,
+          SchemaHandling.ForceCreateAndContinue,
+        )
+        _ = res1 shouldBe Continue
         _ <- dao.transact(sql"INSERT INTO json_api_schema_version(version) VALUES(0)".update.run)
-        _ <- dao.transact(ContractDao.initialize(checkIfExists = true))
+        res2 <- SchemaHandlingResult.fromSchemaHandling(
+          dao,
+          SchemaHandling.CheckAndTerminateIfWrong,
+        )
+        _ = res2 shouldBe Continue
         versions <- dao.transact(sql"SELECT version FROM json_api_schema_version".query[Int].nel)
       } yield {
         Set.from(versions.toList) should not be Set(jdbcDriver.queries.schemaVersion)

diff --git a/ledger-service/http-json/src/main/scala/com/digitalasset/http/Main.scala b/ledger-service/http-json/src/main/scala/com/digitalasset/http/Main.scala
@@ -109,21 +109,19 @@ object Main {
     val contractDao = config.jdbcConfig.map(c => ContractDao(c.driver, c.url, c.user, c.password))
 
     (contractDao, config.jdbcConfig) match {
-      case (Some(dao), Some(c)) if c.createSchema =>
-        logger.info("Creating DB schema...")
-        import dao.{logHandler, jdbcDriver}
-        Try(dao.transact(ContractDao.initialize(c.checkIfExists)).unsafeRunSync()) match {
-          case Success(()) =>
-            logger.info("DB schema created...")
-            if (!c.continueAfterSchemaCreation) {
-              logger.info("Terminating process...")
-              terminate()
-              System.exit(ErrorCodes.Ok)
-            }
+      case (Some(dao), Some(c)) =>
+        def terminateProcess(): Unit = {
+          logger.info("Terminating process...")
+          terminate()
+          System.exit(ErrorCodes.Ok)
+        }
+        Try(SchemaHandlingResult.fromSchemaHandling(dao, c.schemaHandling).unsafeRunSync()) match {
+          case Success(SchemaHandlingResult.Terminate) =>
+            terminateProcess()
+          case Success(SchemaHandlingResult.Continue) => ()
           case Failure(e) =>
-            logger.error("Failed creating DB schema", e)
-            terminate()
-            System.exit(ErrorCodes.StartupError)
+            logger.error("Failed processing the schema handling", e)
+            terminateProcess()
         }
       case _ =>
     }

diff --git a/ledger-service/http-json/src/main/scala/com/digitalasset/http/SchemaHandlingResult.scala b/ledger-service/http-json/src/main/scala/com/digitalasset/http/SchemaHandlingResult.scala
@@ -0,0 +1,61 @@
+package com.daml.http
+
+import com.daml.http.dbbackend.ContractDao
+import com.daml.logging.{ContextualizedLogger, LoggingContextOf}
+import cats.effect.IO
+import com.daml.http.util.Logging.InstanceUUID
+
+sealed trait SchemaHandlingResult
+object SchemaHandlingResult {
+  case object Terminate extends SchemaHandlingResult
+  case object Continue extends SchemaHandlingResult
+
+  def fromBool(shouldTerminate: Boolean): SchemaHandlingResult =
+    if (shouldTerminate) Terminate else Continue
+
+  private[this] val logger = ContextualizedLogger.get(getClass)
+
+  def fromSchemaHandling(dao: ContractDao, schemaHandling: SchemaHandling)(implicit
+      lc: LoggingContextOf[InstanceUUID]
+  ): IO[SchemaHandlingResult] = {
+    import dao.{logHandler, jdbcDriver}
+    def terminate: IO[SchemaHandlingResult] = IO.pure(Terminate)
+    def reinit(shouldTerminate: Boolean): IO[SchemaHandlingResult] = {
+      logger.info("Creating DB schema...")
+      dao.transact(ContractDao.initialize).map { _ =>
+        logger.info("DB schema created...")
+        SchemaHandlingResult.fromBool(shouldTerminate)
+      }
+    }
+    def checkVersion(shouldTerminate: Boolean): IO[SchemaHandlingResult] = {
+      logger.info("Checking for existing schema")
+      dao.transact(jdbcDriver.queries.version()).flatMap {
+        case None =>
+          logger.info("No schema version found")
+          if (shouldTerminate) terminate
+          else reinit(shouldTerminate = false)
+        case Some(version) =>
+          logger.info(s"DB schema version $version found")
+          if (version < jdbcDriver.queries.schemaVersion) {
+            logger.info("DB schema version is not up to date")
+            if (shouldTerminate) terminate
+            else {
+              logger.info(s"Re-initializing with version ${jdbcDriver.queries.schemaVersion}")
+              reinit(shouldTerminate = false)
+            }
+          } else {
+            logger.info("DB schema is up-to-date, continuing startup")
+            IO.pure(Continue)
+          }
+      }
+    }
+    schemaHandling match {
+      case SchemaHandling.ForceCreateAndTerminate => reinit(shouldTerminate = true)
+      case SchemaHandling.ForceCreateAndContinue => reinit(shouldTerminate = false)
+      case SchemaHandling.CheckAndTerminateIfWrong =>
+        checkVersion(shouldTerminate = true)
+      case SchemaHandling.CreateOrUpdateAndContinue =>
+        checkVersion(shouldTerminate = false)
+    }
+  }
+}
diff --git a/ledger-service/http-json/src/main/scala/com/digitalasset/http/dbbackend/ContractDao.scala b/ledger-service/http-json/src/main/scala/com/digitalasset/http/dbbackend/ContractDao.scala
@@ -52,19 +52,10 @@ object ContractDao {
     new ContractDao(Connection.connect(jdbcDriver, jdbcUrl, username, password)(cs))
   }
 
-  def initialize(checkIfExists: Boolean)(implicit
+  def initialize(implicit
       log: LogHandler,
       sjd: SupportedJdbcDriver,
-  ): ConnectionIO[Unit] = {
-    val reInit = sjd.queries.dropAllTablesIfExist *> sjd.queries.initDatabase
-    if (checkIfExists)
-      sjd.queries
-        .version()
-        .flatMap(optVersion =>
-          if (optVersion contains sjd.queries.schemaVersion) fconn.unit else reInit
-        )
-    else reInit
-  }
+  ): ConnectionIO[Unit] = sjd.queries.dropAllTablesIfExist *> sjd.queries.initDatabase
 
   def lastOffset(parties: OneAnd[Set, domain.Party], templateId: domain.TemplateId.RequiredPkg)(
       implicit

diff --git a/ledger-service/http-json/src/test/scala/com/digitalasset/http/CliSpec.scala b/ledger-service/http-json/src/test/scala/com/digitalasset/http/CliSpec.scala
@@ -19,7 +19,6 @@ final class CliSpec extends AnyFreeSpec with Matchers {
     "jdbc:postgresql://localhost:5432/test?&ssl=true",
     "postgres",
     "password",
-    false,
   )
   val jdbcConfigString =
     "driver=org.postgresql.Driver,url=jdbc:postgresql://localhost:5432/test?&ssl=true,user=postgres,password=password,createSchema=false"