feat: PostgreSQL support

.env

@@ -1,6 +1,9 @@
 RUST_LOG=trace,hyper=info,tracing=info,reqwest=info
 
 # The database that should be used
+# SQLite and PostgreSQL are supported:
+# - sqlite://./db.sqlite
+# - postgresql://boltz:[email protected]:5432/covclaim
 DATABASE_URL=sqlite://./db.sqlite
 
 # When finding a lockup transaction, how many seconds to wait before broadcasting the covenant claim (0 for instantly)

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "covclaim"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2021"
 build = "build.rs"
 
@@ -14,7 +14,7 @@ panic = "abort"
 [dependencies]
 tokio = { version = "1.38.0", features = ["full"] }
 axum = "0.7.5"
-diesel = { version = "2.2.1", features = ["sqlite", "r2d2", "chrono"] }
+diesel = { version = "2.2.1", features = ["sqlite", "postgres", "r2d2", "chrono"] }
 diesel_migrations = "2.2.0"
 dotenvy = "0.15.7"
 env_logger = "0.11.3"

migrations_postgres/2024-02-14-014006_setup_schema/down.sql

@@ -0,0 +1,2 @@
+DROP TABLE parameters;
+DROP TABLE pending_covenants;

migrations_postgres/2024-02-14-014006_setup_schema/up.sql

@@ -0,0 +1,19 @@
+CREATE TABLE parameters (
+    name VARCHAR PRIMARY KEY NOT NULL,
+    value VARCHAR NOT NULL
+);
+
+CREATE TABLE pending_covenants (
+    output_script BYTEA PRIMARY KEY NOT NULL,
+    status INT NOT NULL,
+    internal_key BYTEA NOT NULL,
+    preimage BYTEA NOT NULL,
+    swap_tree VARCHAR NOT NULL,
+    address BYTEA NOT NULL,
+    blinding_key BYTEA,
+    tx_id BYTEA,
+    tx_time TIMESTAMP,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX pending_covenants_status_idx ON pending_covenants (status);

src/db/helpers.rs

@@ -9,18 +9,40 @@ use crate::db::schema::pending_covenants;
 
 const BLOCK_HEIGHT_NAME: &str = "block_height";
 
-pub fn upsert_block_height(con: db::Pool, height: u64) -> QueryResult<usize> {
+pub fn upsert_block_height(con: db::Pool, height: u64) -> Result<(), diesel::result::Error> {
     let values = Parameter {
         name: BLOCK_HEIGHT_NAME.to_string(),
         value: height.to_string(),
     };
 
-    insert_into(parameters::dsl::parameters)
-        .values(&values)
-        .on_conflict(parameters::dsl::name)
-        .do_update()
-        .set(parameters::dsl::value.eq(height.to_string()))
-        .execute(&mut con.get().unwrap())
+    match parameters::dsl::parameters
+        .select(Parameter::as_select())
+        .filter(parameters::dsl::name.eq(BLOCK_HEIGHT_NAME.to_string()))
+        .limit(1)
+        .load(&mut con.get().unwrap())
+    {
+        Ok(res) => {
+            if res.is_empty() {
+                match insert_into(parameters::dsl::parameters)
+                    .values(&values)
+                    .execute(&mut con.get().unwrap())
+                {
+                    Ok(_) => Ok(()),
+                    Err(err) => Err(err),
+                }
+            } else {
+                match update(parameters::dsl::parameters)
+                    .filter(parameters::dsl::name.eq(BLOCK_HEIGHT_NAME.to_string()))
+                    .set((parameters::dsl::value.eq(height.to_string()),))
+                    .execute(&mut con.get().unwrap())
+                {
+                    Ok(_) => Ok(()),
+                    Err(err) => Err(err),
+                }
+            }
+        }
+        Err(err) => Err(err),
+    }
 }
 
 pub fn get_block_height(con: db::Pool) -> Option<u64> {

src/db/mod.rs

@@ -1,31 +1,58 @@
+use std::error::Error;
+
+use diesel::prelude::*;
 use diesel::r2d2::ConnectionManager;
-use diesel::sqlite::Sqlite;
-use diesel::{Connection, SqliteConnection};
+use diesel::Connection;
 use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
-use std::error::Error;
+use log::info;
 
 pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("./migrations");
+pub const MIGRATIONS_POSTGRES: EmbeddedMigrations = embed_migrations!("./migrations_postgres");
 
 pub mod helpers;
 
 pub mod models;
 mod schema;
 
-pub type DatabaseConnection = SqliteConnection;
-pub type Pool = r2d2::Pool<ConnectionManager<DatabaseConnection>>;
+#[derive(diesel::MultiConnection)]
+pub enum AnyConnection {
+    Postgresql(PgConnection),
+    Sqlite(SqliteConnection),
+}
 
-pub fn establish_connection(url: &str) -> Result<Pool, Box<dyn Error + Send + Sync>> {
-    run_migrations(&mut SqliteConnection::establish(url)?)?;
+pub type Pool = r2d2::Pool<ConnectionManager<AnyConnection>>;
 
-    let manager = ConnectionManager::<DatabaseConnection>::new(url);
+pub fn establish_connection(url: &str) -> Result<Pool, Box<dyn Error + Send + Sync>> {
+    info!(
+        "Using {} database",
+        if is_postgres_connection_url(url) {
+            "PostgreSQL"
+        } else {
+            "SQLite"
+        }
+    );
+
+    let manager = ConnectionManager::new(url);
     let pool = Pool::builder().build(manager)?;
 
+    run_migrations(is_postgres_connection_url(url), &pool)?;
+
     Ok(pool)
 }
 
 fn run_migrations(
-    connection: &mut impl MigrationHarness<Sqlite>,
+    is_postgres: bool,
+    pool: &Pool,
 ) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
-    connection.run_pending_migrations(MIGRATIONS)?;
+    let mut con = pool.get()?;
+    con.run_pending_migrations(if is_postgres {
+        MIGRATIONS_POSTGRES
+    } else {
+        MIGRATIONS
+    })?;
     Ok(())
 }
+
+fn is_postgres_connection_url(url: &str) -> bool {
+    url.starts_with("postgresql")
+}

src/main.rs

@@ -24,12 +24,20 @@ async fn main() {
     env_logger::init();
 
     info!(
-        "Starting covclaim v{}-{}",
+        "Starting {} v{}-{}{}",
+        built_info::PKG_NAME,
         built_info::PKG_VERSION,
-        built_info::GIT_VERSION.unwrap_or("")
+        built_info::GIT_VERSION.unwrap_or(""),
+        if built_info::GIT_DIRTY.unwrap_or(false) {
+            "-dirty"
+        } else {
+            ""
+        }
     );
+
     debug!(
-        "Compiled with {} for {}",
+        "Compiled {} with {} for {}",
+        built_info::PROFILE,
         built_info::RUSTC_VERSION,
         built_info::TARGET
     );
@@ -101,8 +109,8 @@ async fn get_chain_backend() -> Arc<Box<dyn ChainBackend + Send + Sync>> {
                     .expect("ELEMENTS_PORT invalid"),
                 env::var("ELEMENTS_COOKIE").expect("ELEMENTS_COOKIE must be set"),
             )
-            .connect()
-            .await
+                .connect()
+                .await
             {
                 Ok(client) => Box::new(client),
                 Err(err) => {