Use sqlx inplace of rusqlite

teos/Cargo.toml

@@ -14,6 +14,12 @@ path = "src/cli.rs"
 name = "teosd"
 path = "src/main.rs"
 
+[features]
+# By default, enable both SQLite snd PostgreSQL in the output binary.
+default = ["sqlite", "postgres"]
+sqlite = ["sqlx/sqlite"]
+postgres = ["sqlx/postgres"]
+
 [dependencies]
 # General
 hex = { version = "0.4.3", features = [ "serde" ] }
@@ -22,7 +28,7 @@ log = "0.4"
 prost = "0.9"
 rcgen = { version = "0.8", features = ["pem", "x509-parser"] }
 rusqlite = { version = "0.29.0", features = [ "bundled", "limits" ] }
-sqlx = { version = "0.7", features = ["runtime-tokio", "tls-native-tls"] }
+sqlx = { version = "0.7", features = ["runtime-tokio", "tls-native-tls", "migrate", "any"] }
 serde = "1.0.130"
 serde_json = "1.0"
 simple_logger = "2.1.0"

teos/build.rs

@@ -1,4 +1,6 @@
-fn main() -> Result<(), Box<dyn std::error::Error>> {
+fn main() {
+    // trigger recompilation when a new migration is added without a change in the source code.
+    println!("cargo:rerun-if-changed=migrations");
     tonic_build::configure()
         .extern_path(".common.teos.v2", "::teos-common::protos")
         .type_attribute(".", "#[derive(serde::Serialize, serde::Deserialize)]")
@@ -23,7 +25,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
                 "proto/teos/v2/user.proto",
             ],
             &["proto/teos/v2", "../teos-common/proto/"],
-        )?;
-
-    Ok(())
+        )
+        .unwrap();
 }

teos/migrations/README.md

@@ -0,0 +1,10 @@
+# Structure:
+- `postgres`: Contains `.sql` migrations for postgres databases.
+- `sqlite`: Contains `.sql` migrations for sqlite databases.
+
+
+# Migrations Extra Documentation (`migrations/*/*.md`):
+
+Migrations cannot be edited once applied to the database. Thus, writing/editing any comments or explanations in the `.sql` files would break the tower for users who have applied those migrations.
+
+Any additional comments that we need to add after a migration has been applied should be in `MID_MNAME.md` instead.

teos/migrations/postgres/000_init.sql

@@ -0,0 +1,51 @@
+-- INT is 4 bytes signed integer (i32) in PostgreSQL.
+-- Many fields in here map to (u32)s in Rust, which has double the capacity of (i32)s on the positive side.
+-- Some database calls might break because of this.
+-- A solution for this could be either one of:
+--   1- Find a one to one mapping between the Rust (u32)s and PostgreSQL's (i32)s since they are essentially the same size.
+--   2- Use PostgreSQL's BIGINT which is equivalent to an i64.
+
+CREATE TABLE IF NOT EXISTS users (
+    user_id BYTEA PRIMARY KEY,
+    available_slots BIGINT NOT NULL,
+    subscription_start BIGINT NOT NULL,
+    subscription_expiry BIGINT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS appointments (
+    UUID BYTEA PRIMARY KEY,
+    locator BYTEA NOT NULL,
+    encrypted_blob BYTEA NOT NULL,
+    to_self_delay BIGINT NOT NULL,
+    user_signature TEXT NOT NULL,
+    start_block BIGINT NOT NULL,
+    user_id BYTEA NOT NULL,
+    FOREIGN KEY(user_id)
+        REFERENCES users(user_id)
+        ON DELETE CASCADE
+);
+
+CREATE TABLE IF NOT EXISTS trackers (
+    UUID BYTEA PRIMARY KEY,
+    dispute_tx BYTEA NOT NULL,
+    penalty_tx BYTEA NOT NULL,
+    height BIGINT NOT NULL,
+    confirmed BIGINT NOT NULL,
+    FOREIGN KEY(UUID)
+        REFERENCES appointments(UUID)
+        ON DELETE CASCADE
+);
+
+CREATE TABLE IF NOT EXISTS last_known_block (
+    id INT PRIMARY KEY,
+    block_hash BYTEA NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS keys (
+    id SERIAL PRIMARY KEY,
+    secret_key TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS locators_index ON appointments (
+    locator
+);

teos/migrations/sqlite/000_init.sql

@@ -0,0 +1,40 @@
+CREATE TABLE IF NOT EXISTS users (
+    user_id INT PRIMARY KEY,
+    available_slots INT NOT NULL,
+    subscription_start INT NOT NULL,
+    subscription_expiry INT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS appointments (
+    UUID INT PRIMARY KEY,
+    locator INT NOT NULL,
+    encrypted_blob BLOB NOT NULL,
+    to_self_delay INT NOT NULL,
+    user_signature BLOB NOT NULL,
+    start_block INT NOT NULL,
+    user_id INT NOT NULL,
+    FOREIGN KEY(user_id)
+        REFERENCES users(user_id)
+        ON DELETE CASCADE
+);
+
+CREATE TABLE IF NOT EXISTS trackers (
+    UUID INT PRIMARY KEY,
+    dispute_tx BLOB NOT NULL,
+    penalty_tx BLOB NOT NULL,
+    height INT NOT NULL,
+    confirmed BOOL NOT NULL,
+    FOREIGN KEY(UUID)
+        REFERENCES appointments(UUID)
+        ON DELETE CASCADE
+);
+
+CREATE TABLE IF NOT EXISTS last_known_block (
+    id INT PRIMARY KEY,
+    block_hash INT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS keys (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    key INT NOT NULL
+);

teos/migrations/sqlite/001_datatypes_correction.sql

@@ -0,0 +1,51 @@
+-- Change `user_id` from INT to BLOB.
+CREATE TABLE tmp_users (
+    user_id BLOB PRIMARY KEY,
+    available_slots INT NOT NULL,
+    subscription_start INT NOT NULL,
+    subscription_expiry INT NOT NULL
+);
+INSERT INTO tmp_users SELECT * FROM users;
+-- We couldn't drop `users` before copying `appointments`, as the former will cascade delete the latter.
+-- Same for `trackers` and `appointments`.
+-- DROP TABLE users;
+
+-- Change `UUID` & `locator` & `user_id` from INT to BLOB.
+-- Change `user_signature` from BLOB to TEXT.
+CREATE TABLE tmp_appointments (
+    UUID BLOB PRIMARY KEY,
+    locator BLOB NOT NULL,
+    encrypted_blob BLOB NOT NULL,
+    to_self_delay INT NOT NULL,
+    user_signature TEXT NOT NULL,
+    start_block INT NOT NULL,
+    user_id BLOB NOT NULL,
+    FOREIGN KEY(user_id)
+        REFERENCES tmp_users(user_id)
+        ON DELETE CASCADE
+);
+INSERT INTO tmp_appointments SELECT * FROM appointments;
+
+-- Change `UUID` from INT to BLOB.
+-- Change `confirmed` from BOOL to INT (due to https://github.com/launchbadge/sqlx/issues/2657).
+CREATE TABLE tmp_trackers (
+    UUID BLOB PRIMARY KEY,
+    dispute_tx BLOB NOT NULL,
+    penalty_tx BLOB NOT NULL,
+    height INT NOT NULL,
+    confirmed INT NOT NULL,
+    FOREIGN KEY(UUID)
+        REFERENCES tmp_appointments(UUID)
+        ON DELETE CASCADE
+);
+INSERT INTO tmp_trackers SELECT * FROM trackers;
+
+-- We can drop these now after all the data has been copied.
+DROP TABLE users;
+DROP TABLE appointments;
+DROP TABLE trackers;
+
+-- Foreign key references are automatically adjusted (tmp_* -> *).
+ALTER TABLE tmp_users RENAME TO users;
+ALTER TABLE tmp_appointments RENAME TO appointments;
+ALTER TABLE tmp_trackers RENAME TO trackers;

teos/migrations/sqlite/002_more_datatypes_corrections.sql

@@ -0,0 +1,17 @@
+-- Change `block_hash` from INT to BLOB.
+CREATE TABLE tmp_last_known_block (
+    id INT PRIMARY KEY,
+    block_hash BLOB NOT NULL
+);
+INSERT INTO tmp_last_known_block SELECT * FROM last_known_block;
+DROP TABLE last_known_block;
+ALTER TABLE tmp_last_known_block RENAME TO last_known_block;
+
+-- Change `key` from INT to TEXT.
+CREATE TABLE tmp_keys (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    key TEXT NOT NULL
+);
+INSERT INTO tmp_keys SELECT * FROM keys;
+DROP TABLE keys;
+ALTER TABLE tmp_keys RENAME TO keys;

teos/migrations/sqlite/003_keys_rename.sql

@@ -0,0 +1,2 @@
+-- Rename `key` to `secret_key` as the word `key` is reserved in some databases.
+ALTER TABLE keys RENAME key TO secret_key;

teos/migrations/sqlite/004_locator_index.sql

@@ -0,0 +1,5 @@
+-- This index greatly enhances the performance of locator based selection queries:
+-- "SELECT ... FROM appointments WHERE locator = ..."
+CREATE INDEX IF NOT EXISTS locators_index ON appointments (
+    locator
+);

teos/src/listener_actor.rs → teos/src/async_listener.rs

@@ -1,3 +1,6 @@
+//! Contains the [AsyncListen] trait that's analogous to the [chain::Listen] from LDK but runs
+//! inside an asynchronous context.
+
 use crate::dbm::DBM;
 
 use std::marker::{Send, Sync};
@@ -7,6 +10,7 @@ use bitcoin::{Block, BlockHeader};
 use lightning::chain;
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
 
+/// A trait similar to LDK's [chain::Listen] but runs asynchronously.
 #[tonic::async_trait]
 pub trait AsyncListen: Send + Sync {
     async fn block_connected(&self, block: &Block, height: u32);
@@ -43,24 +47,31 @@ enum BlockListenerAction {
     BlockDisconnected(BlockHeader, u32),
 }
 
-pub struct AsyncBlockListener<L> {
+/// A helper struct that wraps a listener that implements [AsyncListen] and feeds it connected and disconnected
+/// blocks received from [UnboundedReceiver] in the background.
+pub struct AsyncBlockListener<L: AsyncListen> {
     listener: L,
     dbm: Arc<Mutex<DBM>>,
     rx: UnboundedReceiver<BlockListenerAction>,
 }
 
-impl<L> AsyncBlockListener<L>
-where
-    L: AsyncListen + 'static,
-{
-    pub fn new(listener: L, dbm: Arc<Mutex<DBM>>) -> SyncBlockListener {
+impl<L: AsyncListen + 'static> AsyncBlockListener<L> {
+    /// Takes a `listener` that implements [AsyncListen] and returns a listener that implements [chain::Listen].
+    ///
+    /// These two listeners are connected. That is, blocks connected-to/disconnected-from the [chain::Listen]
+    /// listener are forwarded to the [AsyncListen] listener.
+    ///
+    /// The [AsyncListen] listener will be actively listening for actions in a background tokio task.
+    pub fn wrap_listener(listener: L, dbm: Arc<Mutex<DBM>>) -> SyncBlockListener {
         let (tx, rx) = unbounded_channel();
         let actor = AsyncBlockListener { listener, dbm, rx };
         actor.run_actor_in_bg();
         SyncBlockListener { tx }
     }
 
-    fn run_actor_in_bg(mut self: Self) {
+    /// Spawns a forever living task that listens for [BlockListenerAction] and feeds them to the
+    /// listener in an asynchronous context.
+    fn run_actor_in_bg(mut self) {
         tokio::spawn(async move {
             while let Some(action) = self.rx.recv().await {
                 match action {
@@ -82,7 +93,8 @@ where
     }
 }
 
-#[derive(Debug)]
+/// A block listener that implements the sync [chain::Listen] trait. All it does is forward the blocks received
+/// another (async) block listener through an [UnboundedSender].
 pub struct SyncBlockListener {
     tx: UnboundedSender<BlockListenerAction>,
 }
@@ -107,7 +119,7 @@ impl chain::Listen for SyncBlockListener {
         height: u32,
     ) {
         let block = Block {
-            header: header.clone(),
+            header: *header,
             txdata: txdata.iter().map(|&(_, tx)| tx.clone()).collect(),
         };
         self.tx

teos/src/chain_monitor.rs

@@ -199,14 +199,8 @@ mod tests {
         let spv_client = SpvClient::new(old_tip, poller, cache, &listener);
         let bitcoind_reachable = Arc::new((Mutex::new(true), Condvar::new()));
 
-        let mut cm = ChainMonitor::new(
-            spv_client,
-            old_tip,
-            1,
-            shutdown_signal,
-            bitcoind_reachable,
-        )
-        .await;
+        let mut cm =
+            ChainMonitor::new(spv_client, old_tip, 1, shutdown_signal, bitcoind_reachable).await;
 
         // If a new (best) block gets mined, it should be connected
         cm.poll_best_tip().await;
@@ -236,14 +230,8 @@ mod tests {
         let spv_client = SpvClient::new(best_tip, poller, cache, &listener);
         let bitcoind_reachable = Arc::new((Mutex::new(true), Condvar::new()));
 
-        let mut cm = ChainMonitor::new(
-            spv_client,
-            best_tip,
-            1,
-            shutdown_signal,
-            bitcoind_reachable,
-        )
-        .await;
+        let mut cm =
+            ChainMonitor::new(spv_client, best_tip, 1, shutdown_signal, bitcoind_reachable).await;
 
         // If a new (worse, just one) block gets mined, nothing gets connected nor disconnected
         cm.poll_best_tip().await;
@@ -272,14 +260,8 @@ mod tests {
         let spv_client = SpvClient::new(old_best, poller, cache, &listener);
         let bitcoind_reachable = Arc::new((Mutex::new(true), Condvar::new()));
 
-        let mut cm = ChainMonitor::new(
-            spv_client,
-            old_best,
-            1,
-            shutdown_signal,
-            bitcoind_reachable,
-        )
-        .await;
+        let mut cm =
+            ChainMonitor::new(spv_client, old_best, 1, shutdown_signal, bitcoind_reachable).await;
 
         // If a a reorg is found (tip is disconnected and a new best is found), both data should be connected and disconnected
         cm.poll_best_tip().await;