Check applied migration

12 files changed, 74 insertions, 58 deletions

diff --git a/bank/src/main/kotlin/tech/libeufin/bank/Main.kt b/bank/src/main/kotlin/tech/libeufin/bank/Main.kt
index 8dafbce7..db2e6d8d 100644
--- a/bank/src/main/kotlin/tech/libeufin/bank/Main.kt
+++ b/bank/src/main/kotlin/tech/libeufin/bank/Main.kt
@@ -253,7 +253,7 @@ fun Application.corebankWebApp(db: Database, ctx: BankConfig) {
 
 class BankDbInit : CliktCommand("Initialize the libeufin-bank database", name = "dbinit") {
     private val common by CommonOption()
-    private val requestReset by option(
+    private val reset by option(
         "--reset", "-r",
         help = "Reset database (DANGEROUS: All existing data is lost)"
     ).flag()
@@ -262,14 +262,8 @@ class BankDbInit : CliktCommand("Initialize the libeufin-bank database", name =
         val config = talerConfig(common.config)
         val cfg = config.loadDbConfig()
         val ctx = config.loadBankConfig()
-        Database(cfg.dbConnStr, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
-            db.conn { conn ->
-                if (requestReset) {
-                    resetDatabaseTables(conn, cfg, sqlFilePrefix = "libeufin-bank")
-                }
-                initializeDatabaseTables(conn, cfg, sqlFilePrefix = "libeufin-bank")
-            }
-            
+        pgDataSource(cfg.dbConnStr).dbInit(cfg, "libeufin-bank", reset)
+        Database(cfg, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
             // Create admin account if missing
             val res = createAdminAccount(db, ctx) // logs provided by the helper
             when (res) {
@@ -292,7 +286,7 @@ class ServeBank : CliktCommand("Run libeufin-bank HTTP server", name = "serve")
         val ctx = cfg.loadBankConfig()
         val dbCfg = cfg.loadDbConfig()
         val serverCfg = cfg.loadServerConfig()
-        Database(dbCfg.dbConnStr, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
+        Database(dbCfg, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
             if (ctx.allowConversion) {
                 logger.info("Ensure exchange account exists")
                 val info = db.account.bankInfo("exchange", ctx.payto)
@@ -351,7 +345,7 @@ class ChangePw : CliktCommand("Change account password", name = "passwd") {
         val cfg = talerConfig(common.config)
         val ctx = cfg.loadBankConfig() 
         val dbCfg = cfg.loadDbConfig()
-        Database(dbCfg.dbConnStr, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
+        Database(dbCfg, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
             val res = db.account.reconfigPassword(username, password, null, true)
             when (res) {
                 AccountPatchAuthResult.UnknownAccount ->
@@ -395,7 +389,7 @@ class EditAccount : CliktCommand(
         val cfg = talerConfig(common.config)
         val ctx = cfg.loadBankConfig() 
         val dbCfg = cfg.loadDbConfig()
-        Database(dbCfg.dbConnStr, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
+        Database(dbCfg, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
             val req = AccountReconfiguration(
                 name = name,
                 is_taler_exchange = exchange,
@@ -472,7 +466,7 @@ class CreateAccount : CliktCommand(
         val ctx = cfg.loadBankConfig() 
         val dbCfg = cfg.loadDbConfig()
 
-        Database(dbCfg.dbConnStr, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
+        Database(dbCfg, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
             val req = json ?: options?.run {
                 RegisterAccountRequest(
                     username = username,
@@ -519,7 +513,7 @@ class GC : CliktCommand(
         val ctx = cfg.loadBankConfig() 
         val dbCfg = cfg.loadDbConfig()
 
-        Database(dbCfg.dbConnStr, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
+        Database(dbCfg, ctx.regionalCurrency, ctx.fiatCurrency).use { db ->
             logger.info("Run garbage collection")
             db.gc.collect(Instant.now(), ctx.gcAbortAfter, ctx.gcCleanAfter, ctx.gcDeleteAfter)
         }
diff --git a/bank/src/main/kotlin/tech/libeufin/bank/db/Database.kt b/bank/src/main/kotlin/tech/libeufin/bank/db/Database.kt
index b021ff6c..3f47bd06 100644
--- a/bank/src/main/kotlin/tech/libeufin/bank/db/Database.kt
+++ b/bank/src/main/kotlin/tech/libeufin/bank/db/Database.kt
@@ -37,7 +37,7 @@ import kotlin.math.abs
 
 private val logger: Logger = LoggerFactory.getLogger("libeufin-bank-db")
 
-class Database(dbConfig: String, internal val bankCurrency: String, internal val fiatCurrency: String?): DbPool(dbConfig, "libeufin_bank") {
+class Database(dbConfig: DatabaseConfig, internal val bankCurrency: String, internal val fiatCurrency: String?): DbPool(dbConfig, "libeufin-bank") {
     internal val notifWatcher: NotificationWatcher = NotificationWatcher(pgSource)
 
     val cashout = CashoutDAO(this)
diff --git a/bank/src/test/kotlin/helpers.kt b/bank/src/test/kotlin/helpers.kt
index 06db3b8e..1a1f5b19 100644
--- a/bank/src/test/kotlin/helpers.kt
+++ b/bank/src/test/kotlin/helpers.kt
@@ -62,12 +62,12 @@ fun setup(
     val config = talerConfig(Path("conf/$conf"))
     val dbCfg = config.loadDbConfig()
     val ctx = config.loadBankConfig()
-    Database(dbCfg.dbConnStr, ctx.regionalCurrency, ctx.fiatCurrency).use {
+    pgDataSource(dbCfg.dbConnStr).run {
+        dbInit(dbCfg, "libeufin-nexus", true)
+        dbInit(dbCfg, "libeufin-bank", true)
+    }
+    Database(dbCfg, ctx.regionalCurrency, ctx.fiatCurrency).use {
         it.conn { conn ->
-            resetDatabaseTables(conn, dbCfg, "libeufin-nexus")
-            initializeDatabaseTables(conn, dbCfg, "libeufin-nexus")
-            resetDatabaseTables(conn, dbCfg, "libeufin-bank")
-            initializeDatabaseTables(conn, dbCfg, "libeufin-bank")
             val sqlProcedures = Path("${dbCfg.sqlDir}/libeufin-conversion-setup.sql")
             conn.execSQLUpdate(sqlProcedures.readText())
         }
diff --git a/common/src/main/kotlin/db/DbPool.kt b/common/src/main/kotlin/db/DbPool.kt
index 4b6944cc..f003e1d2 100644
--- a/common/src/main/kotlin/db/DbPool.kt
+++ b/common/src/main/kotlin/db/DbPool.kt
@@ -28,14 +28,14 @@ import com.zaxxer.hikari.HikariDataSource
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
 
-open class DbPool(cfg: String, schema: String) : java.io.Closeable {
-    val pgSource = pgDataSource(cfg)
+open class DbPool(cfg: DatabaseConfig, schema: String) : java.io.Closeable {
+    val pgSource = pgDataSource(cfg.dbConnStr)
     private val pool: HikariDataSource
 
     init {
         val config = HikariConfig()
         config.dataSource = pgSource
-        config.schema = schema
+        config.schema = schema.replace("-", "_")
         config.transactionIsolation = "TRANSACTION_SERIALIZABLE"
         pool = HikariDataSource(config)
         pool.connection.use { con ->
@@ -45,6 +45,7 @@ open class DbPool(cfg: String, schema: String) : java.io.Closeable {
             if (majorVersion < MIN_VERSION) {
                 throw Exception("postgres version must be at least $MIN_VERSION.0 got $majorVersion.$minorVersion")
             }
+            checkMigrations(con, cfg, schema)
         }
     }
 
diff --git a/common/src/main/kotlin/db/schema.kt b/common/src/main/kotlin/db/schema.kt
index af589cbb..176c04ee 100644
--- a/common/src/main/kotlin/db/schema.kt
+++ b/common/src/main/kotlin/db/schema.kt
@@ -21,6 +21,8 @@ package tech.libeufin.common.db
 
 import tech.libeufin.common.*
 import org.postgresql.jdbc.PgConnection
+import org.postgresql.ds.PGSimpleDataSource
+import java.sql.Connection
 import kotlin.io.path.Path
 import kotlin.io.path.exists
 import kotlin.io.path.readText
@@ -31,7 +33,7 @@ import kotlin.io.path.readText
  * @param conn database connection
  * @param cfg database configuration
  */
-fun maybeApplyV(conn: PgConnection, cfg: DatabaseConfig) {
+private fun maybeApplyV(conn: PgConnection, cfg: DatabaseConfig) {
     conn.transaction {
         val checkVSchema = conn.prepareStatement(
             "SELECT schema_name FROM information_schema.schemata WHERE schema_name = '_v'"
@@ -44,32 +46,35 @@ fun maybeApplyV(conn: PgConnection, cfg: DatabaseConfig) {
     }
 }
 
+private fun migrationsPath(sqlFilePrefix: String): Sequence<String> = sequence {
+    for (n in 1..9999) {
+        val padded = n.toString().padStart(4, '0')
+        yield("$sqlFilePrefix-$padded")
+    }
+}
+
 // sqlFilePrefix is, for example, "libeufin-bank" or "libeufin-nexus" (no trailing dash).
-fun initializeDatabaseTables(conn: PgConnection, cfg: DatabaseConfig, sqlFilePrefix: String) {
+private fun initializeDatabaseTables(conn: PgConnection, cfg: DatabaseConfig, sqlFilePrefix: String) {
     logger.info("doing DB initialization, sqldir ${cfg.sqlDir}")
     maybeApplyV(conn, cfg)
     conn.transaction {
-        val checkStmt = conn.prepareStatement("SELECT count(*) as n FROM _v.patches where patch_name = ?")
-
-        for (n in 1..9999) {
-            val numStr = n.toString().padStart(4, '0')
-            val patchName = "$sqlFilePrefix-$numStr"
+        val checkStmt = conn.prepareStatement("SELECT EXISTS(SELECT FROM _v.patches where patch_name = ?)")
 
+        for (patchName in migrationsPath(sqlFilePrefix)) {
             checkStmt.setString(1, patchName)
-            val patchCount = checkStmt.oneOrNull { it.getInt(1) } ?: throw Exception("unable to query patches")
-            if (patchCount >= 1) {
+            val applied = checkStmt.one { it.getBoolean(1) }
+            if (applied) {
                 logger.debug("patch $patchName already applied")
                 continue
             }
 
-            val path = Path("${cfg.sqlDir}/$sqlFilePrefix-$numStr.sql")
+            val path = Path("${cfg.sqlDir}/$patchName.sql")
             if (!path.exists()) {
                 logger.debug("path $path doesn't exist anymore, stopping")
                 break
             }
             logger.info("applying patch $path")
-            val sqlPatchText = path.readText()
-            conn.execSQLUpdate(sqlPatchText)
+            conn.execSQLUpdate(path.readText())
         }
         val sqlProcedures = Path("${cfg.sqlDir}/$sqlFilePrefix-procedures.sql")
         if (!sqlProcedures.exists()) {
@@ -81,9 +86,32 @@ fun initializeDatabaseTables(conn: PgConnection, cfg: DatabaseConfig, sqlFilePre
     }
 }
 
+internal fun checkMigrations(conn: Connection, cfg: DatabaseConfig, sqlFilePrefix: String) {
+    val checkStmt = conn.prepareStatement("SELECT EXISTS(SELECT FROM _v.patches where patch_name = ?)")
+
+    for (patchName in migrationsPath(sqlFilePrefix)) {
+        checkStmt.setString(1, patchName)
+        val path = Path("${cfg.sqlDir}/$patchName.sql")
+        if (!path.exists()) break
+        val applied = checkStmt.one { it.getBoolean(1) }
+        if (!applied) {
+            throw Exception("patch $patchName not applied, run '$sqlFilePrefix dbinit'")
+        }
+    }
+}
+
 // sqlFilePrefix is, for example, "libeufin-bank" or "libeufin-nexus" (no trailing dash).
-fun resetDatabaseTables(conn: PgConnection, cfg: DatabaseConfig, sqlFilePrefix: String) {
+private fun resetDatabaseTables(conn: PgConnection, cfg: DatabaseConfig, sqlFilePrefix: String) {
     logger.info("reset DB, sqldir ${cfg.sqlDir}")
     val sqlDrop = Path("${cfg.sqlDir}/$sqlFilePrefix-drop.sql").readText()
     conn.execSQLUpdate(sqlDrop)
+}
+
+fun PGSimpleDataSource.dbInit(cfg: DatabaseConfig, sqlFilePrefix: String, reset: Boolean) {
+    pgConnection().use { conn ->
+        if (reset) {
+            resetDatabaseTables(conn, cfg, sqlFilePrefix)
+        }
+        initializeDatabaseTables(conn, cfg, sqlFilePrefix)
+    }
 }
\ No newline at end of file
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/DbInit.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/DbInit.kt
index 7120faf2..0128ab85 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/DbInit.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/DbInit.kt
@@ -32,18 +32,13 @@ import tech.libeufin.common.db.*
  */
 class DbInit : CliktCommand("Initialize the libeufin-nexus database", name = "dbinit") {
     private val common by CommonOption()
-    private val requestReset by option(
+    private val reset by option(
         "--reset", "-r",
         help = "Reset database (DANGEROUS: All existing data is lost)"
     ).flag()
 
     override fun run() = cliCmd(logger, common.log) {
         val cfg = loadConfig(common.config).dbConfig()
-        pgDataSource(cfg.dbConnStr).pgConnection().use { conn ->
-            if (requestReset) {
-                resetDatabaseTables(conn, cfg, sqlFilePrefix = "libeufin-nexus")
-            }
-            initializeDatabaseTables(conn, cfg, sqlFilePrefix = "libeufin-nexus")
-        }
+        pgDataSource(cfg.dbConnStr).dbInit(cfg, "libeufin-nexus", reset)
     }
 }
\ No newline at end of file
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt
index fd6be2d9..9e95a52c 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt
@@ -366,7 +366,7 @@ class EbicsFetch: CliktCommand("Fetches EBICS files") {
         val cfg = extractEbicsConfig(common.config)
         val dbCfg = cfg.config.dbConfig()
 
-        Database(dbCfg.dbConnStr).use { db ->
+        Database(dbCfg).use { db ->
             val (clientKeys, bankKeys) = expectFullKeys(cfg)
             val ctx = FetchContext(
                 cfg,
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsSubmit.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsSubmit.kt
index 34d57969..12a72d40 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsSubmit.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsSubmit.kt
@@ -156,7 +156,7 @@ class EbicsSubmit : CliktCommand("Submits any initiated payment found in the dat
             httpClient = HttpClient(),
             fileLogger = FileLogger(ebicsLog)
         )
-        Database(dbCfg.dbConnStr).use { db -> 
+        Database(dbCfg).use { db -> 
             val frequency: Duration = if (transient) {
                 logger.info("Transient mode: submitting what found and returning.")
                 Duration.ZERO
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt
index 496a13cf..7e7a5e5d 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt
@@ -161,7 +161,7 @@ class InitiatePayment: CliktCommand("Initiate an outgoing payment") {
             Base32Crockford.encode(bytes)
         }
 
-        Database(dbCfg.dbConnStr).use { db ->
+        Database(dbCfg).use { db ->
             db.initiated.create(
                 InitiatedPayment(
                     id = -1,
@@ -207,7 +207,7 @@ class FakeIncoming: CliktCommand("Genere a fake incoming payment") {
             Base32Crockford.encode(bytes)
         }
 
-        Database(dbCfg.dbConnStr).use { db ->
+        Database(dbCfg).use { db ->
             ingestIncomingPayment(db, 
                 IncomingPayment(
                     amount = amount,
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/db/Database.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/db/Database.kt
index 069185a9..d76a0405 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/db/Database.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/db/Database.kt
@@ -44,7 +44,7 @@ data class InitiatedPayment(
 /**
  * Collects database connection steps and any operation on the Nexus tables.
  */
-class Database(dbConfig: String): DbPool(dbConfig, "libeufin_nexus") {
+class Database(dbConfig: DatabaseConfig): DbPool(dbConfig, "libeufin_nexus") {
     val payment = PaymentDAO(this)
     val initiated = InitiatedDAO(this)
 }
\ No newline at end of file
diff --git a/nexus/src/test/kotlin/helpers.kt b/nexus/src/test/kotlin/helpers.kt
index f0c0f8f7..2f02f2ec 100644
--- a/nexus/src/test/kotlin/helpers.kt
+++ b/nexus/src/test/kotlin/helpers.kt
@@ -44,11 +44,8 @@ fun setup(
     val config = NEXUS_CONFIG_SOURCE.fromFile(Path("conf/$conf"))
     val dbCfg = config.dbConfig()
     val ctx = NexusConfig(config)
-    Database(dbCfg.dbConnStr).use { 
-        it.conn { conn ->
-            resetDatabaseTables(conn, dbCfg, "libeufin-nexus")
-            initializeDatabaseTables(conn, dbCfg, "libeufin-nexus")
-        }
+    pgDataSource(dbCfg.dbConnStr).dbInit(dbCfg, "libeufin-nexus", true)
+    Database(dbCfg).use {
         lambda(it, ctx)
     }
 }
diff --git a/testbench/src/test/kotlin/IntegrationTest.kt b/testbench/src/test/kotlin/IntegrationTest.kt
index b690462b..a8812c8b 100644
--- a/testbench/src/test/kotlin/IntegrationTest.kt
+++ b/testbench/src/test/kotlin/IntegrationTest.kt
@@ -68,10 +68,11 @@ fun server(lambda: () -> Unit) {
    
 }
 
-fun setup(lambda: suspend (NexusDb) -> Unit) {
+fun setup(conf: String, lambda: suspend (NexusDb) -> Unit) {
     try {
         runBlocking {
-            NexusDb("postgresql:///libeufincheck").use {
+            val cfg = loadConfig(Path(conf)).dbConfig()
+            NexusDb(cfg).use {
                 lambda(it)
             }
         }
@@ -105,7 +106,7 @@ class IntegrationTest {
             bankCmd.run("serve $flags")
         }
         
-        setup { _ ->
+        setup("conf/mini.conf") { _ ->
             // Check bank is running
             client.get("http://0.0.0.0:8080/public-accounts").assertNoContent()
         }
@@ -129,7 +130,7 @@ class IntegrationTest {
             }
         }
 
-        setup { db ->
+        setup("conf/integration.conf")  { db ->
             val userPayTo = IbanPayto.rand()
             val fiatPayTo = IbanPayto.rand()
     
@@ -235,7 +236,7 @@ class IntegrationTest {
             bankCmd.run("serve $flags")
         }
         
-        setup { db -> 
+        setup("conf/integration.conf") { db -> 
             val userPayTo = IbanPayto.rand()
             val fiatPayTo = IbanPayto.rand()