timeseries: add a class to persist start dates in mysql (#253)

Also:
- Secure the db lock before doing schema migration
- Move the lock table creation out of schema migration

core/src/main/scala/com/criteo/cuttle/Database.scala

@@ -79,11 +79,6 @@ private[cuttle] object Database {
 
   val schemaEvolutions = List(
     sql"""
-      CREATE TABLE locks (
-        locked_by       VARCHAR(36) NOT NULL,
-        locked_at       DATETIME NOT NULL
-      ) ENGINE = INNODB;
-
       CREATE TABLE executions (
         id          CHAR(36) NOT NULL,
         job         VARCHAR(1000) NOT NULL,
@@ -119,6 +114,11 @@ private[cuttle] object Database {
 
     // Try to insert our lock at bootstrap
     (for {
+      _ <- sql"""CREATE TABLE IF NOT EXISTS locks (
+          locked_by       VARCHAR(36) NOT NULL,
+          locked_at       DATETIME NOT NULL
+        ) ENGINE = INNODB
+      """.update.run
       locks <- sql"""
           SELECT locked_by, locked_at FROM locks WHERE TIMESTAMPDIFF(MINUTE, locked_at, NOW()) < 5;
         """.query[(String, Instant)].to[List]
@@ -189,9 +189,9 @@ private[cuttle] object Database {
 
   def connect(dbConfig: DatabaseConfig): XA = connections.getOrElseUpdate(
     dbConfig, {
-      val xa = newHikariTransactor(dbConfig)
+      val xa = lockedTransactor(newHikariTransactor(dbConfig))
       doSchemaUpdates.transact(xa).unsafeRunSync
-      lockedTransactor(xa)
+      xa
     }
   )
 }

core/src/main/scala/com/criteo/cuttle/Utils.scala

@@ -19,6 +19,12 @@ import lol.http.{PartialService, Service}
 /** A set of basic utilities useful to write workflows. */
 package object utils {
 
+  /** Get a doobie transactor
+    *
+    * @param config Database configuration
+    */
+  def transactor(config: DatabaseConfig): XA = Database.newHikariTransactor(config)
+
   /** Executes unapplied schema evolutions
     *
     * @param table Name of the table that keeps track of applied schema changes

timeseries/src/main/scala/com/criteo/cuttle/timeseries/contrib/PersistStartDate.scala

@@ -0,0 +1,34 @@
+package com.criteo.cuttle.timeseries.contrib
+
+import doobie.implicits._
+import com.criteo.cuttle._
+import java.time._
+
+class PersistInstant(xa: XA) {
+  def set(id: String, t: Instant): Instant = {
+    sql"REPLACE INTO instant_data VALUES (${id}, ${t})"
+      .update.run.transact(xa).unsafeRunSync
+    t
+  }
+
+  def get(id: String): Option[Instant] =
+    sql"SELECT instant FROM instant_data WHERE id = ${id}"
+      .query[Instant].option.transact(xa).unsafeRunSync
+}
+
+object PersistInstant {
+  private val schemaUpgrades = List(
+    sql"""
+      CREATE TABLE instant_data (
+        id        VARCHAR(1000) NOT NULL,
+        instant   DATETIME NOT NULL,
+        PRIMARY KEY (id)
+      ) ENGINE = INNODB
+    """.update.run
+  )
+
+  def apply(xa: XA): PersistInstant = {
+    utils.updateSchema("instant", schemaUpgrades).transact(xa).unsafeRunSync
+    new PersistInstant(xa)
+  }
+}