moving to tokio runtime, better async handling

3 files changed, 167 insertions, 151 deletions

diff --git a/src/db.rs b/src/db.rs
index a899ac3..99b577e 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -5,7 +5,7 @@ use std::{
     path::{Path, PathBuf},
     time::UNIX_EPOCH,
 };
-use turso::{Connection, params};
+use turso::{Connection, params, transaction::Transaction};
 
 const TABLE_CREATE: &str = "CREATE TABLE IF NOT EXISTS flacs (path TEXT PRIMARY KEY UNIQUE, toencode BOOLEAN NOT NULL, modtime INTEGER)";
 const ADD_FILE: &str = "INSERT INTO flacs (path, toencode, modtime) VALUES (?1, ?2, ?3)";
@@ -15,13 +15,11 @@ const TOENCODE_NUMBER: &str = "SELECT COUNT(*) from flacs WHERE toencode";
 const CHECK_FILE: &str = "SELECT exists(SELECT 1 FROM flacs WHERE path = ?1)";
 const FETCH_FILES: &str = "SELECT path FROM flacs";
 const REMOVE_FILE: &str = "DELETE FROM flacs WHERE path = ?1";
-const DEDUPE_DB: &str =
-    "DELETE FROM flacs WHERE rowid NOT IN (SELECT MAX(rowid) FROM flacs GROUP BY path)";
 const GET_MODTIME: &str = "SELECT modtime FROM flacs WHERE path = ?1";
 
 pub(crate) async fn init_db(path: Option<&PathBuf>) -> Result<turso::Database> {
     let db = if let Some(file) = path {
-        turso::Builder::new_local(file.to_str().unwrap())
+        turso::Builder::new_local(file.canonicalize()?.to_str().unwrap())
             .build()
             .await?
     } else if let Some(base_dir) = BaseDirs::new() {
@@ -37,7 +35,7 @@ pub(crate) async fn init_db(path: Option<&PathBuf>) -> Result<turso::Database> {
     Ok(db)
 }
 
-pub(crate) async fn insert_file(conn: &Connection, filename: &Path) -> Result<()> {
+pub(crate) async fn insert_file<'a>(tx: Transaction<'a>, filename: &Path) -> Result<()> {
     let toencode = !matches!(get_vendor(filename)?.as_str(), CURRENT_VENDOR);
 
     let modtime = filename
@@ -46,33 +44,37 @@ pub(crate) async fn insert_file(conn: &Connection, filename: &Path) -> Result<()
         .duration_since(UNIX_EPOCH)?
         .as_secs();
 
-    conn.execute(
+    tx.execute(
         ADD_FILE,
         params![filename.to_str().unwrap(), toencode, modtime],
     )
     .await?;
 
+    tx.commit().await?;
+
     Ok(())
 }
 
-pub(crate) async fn update_file(conn: &Connection, filename: &Path) -> Result<()> {
+pub(crate) async fn update_file<'a>(tx: Transaction<'a>, filename: &Path) -> Result<()> {
     let modtime = filename
         .metadata()?
         .modified()?
         .duration_since(UNIX_EPOCH)?
         .as_secs();
 
-    conn.execute(
+    tx.execute(
         UPDATE_FILE,
         params![filename.to_str().unwrap(), false, modtime],
     )
     .await?;
 
+    tx.commit().await?;
+
     Ok(())
 }
 
-pub(crate) async fn check_file(conn: &Connection, filename: &Path) -> Result<bool> {
-    Ok(conn
+pub(crate) async fn check_file<'a>(tx: &Transaction<'a>, filename: &Path) -> Result<bool> {
+    Ok(tx
         .query(CHECK_FILE, params!(filename.to_str().unwrap()))
         .await?
         .next()
@@ -81,8 +83,7 @@ pub(crate) async fn check_file(conn: &Connection, filename: &Path) -> Result<boo
         .get::<bool>(0)?)
 }
 
-pub(crate) async fn init_clean_files(conn: &Connection) -> Result<Vec<PathBuf>, turso::Error> {
-    conn.execute(DEDUPE_DB, ()).await?;
+pub(crate) async fn fetch_files(conn: &Connection) -> Result<Vec<PathBuf>, turso::Error> {
     let mut rows = conn.query(FETCH_FILES, ()).await?;
     let mut files = Vec::new();
     while let Ok(Some(row)) = rows.next().await {
@@ -93,9 +94,10 @@ pub(crate) async fn init_clean_files(conn: &Connection) -> Result<Vec<PathBuf>,
     Ok(files)
 }
 
-pub(crate) async fn remove_file(conn: &Connection, filename: &Path) -> Result<()> {
-    conn.execute(REMOVE_FILE, params!(filename.to_str().unwrap()))
+pub(crate) async fn remove_file<'a>(tx: Transaction<'a>, filename: &Path) -> Result<()> {
+    tx.execute(REMOVE_FILE, params!(filename.to_str().unwrap()))
         .await?;
+    tx.commit().await?;
     Ok(())
 }
 
@@ -119,8 +121,8 @@ pub(crate) async fn get_toencode_number(conn: &Connection) -> Result<u64, turso:
         .get::<u64>(0)
 }
 
-pub(crate) async fn get_modtime(conn: &Connection, file: &Path) -> Result<u64> {
-    Ok(conn
+pub(crate) async fn get_modtime<'a>(tx: &Transaction<'a>, file: &Path) -> Result<u64> {
+    Ok(tx
         .query(GET_MODTIME, params![file.to_str().unwrap()])
         .await?
         .next()
@@ -129,8 +131,9 @@ pub(crate) async fn get_modtime(conn: &Connection, file: &Path) -> Result<u64> {
         .get::<u64>(0)?)
 }
 
-pub(crate) async fn vacuum(conn: &Connection) -> Result<()> {
-    conn.execute("VACUUM", ()).await?;
+pub(crate) async fn vacuum<'a>(tx: Transaction<'a>) -> Result<()> {
+    tx.execute("VACUUM", ()).await?;
+    tx.commit().await?;
     Ok(())
 }
 
@@ -138,11 +141,10 @@ pub(crate) async fn vacuum(conn: &Connection) -> Result<()> {
 mod tests {
 
     use super::*;
-    use macro_rules_attribute::apply;
-    use smol_macros::{Executor, test};
+    use turso::transaction::{Transaction, TransactionBehavior::Deferred};
 
-    #[apply(test!)]
-    async fn check_localfiles(ex: &Executor<'_>) {
+    #[tokio::test]
+    async fn check_localfiles() {
         let dbname = PathBuf::from("temp1.db");
         let filenames = [
             "./samples/16bit.flac",
@@ -150,69 +152,67 @@ mod tests {
             "./samples/32bit.flac",
         ];
         let mut counter = 0;
-        ex.spawn(async move {
-            let db = init_db(Some(&dbname)).await.unwrap();
-            let conn = db.connect().unwrap();
-            for file in filenames {
-                let path = PathBuf::from(file);
-                insert_file(&conn, &path).await.unwrap();
-            }
-            let mut returned = conn.query(TOENCODE_PATHS, ()).await.unwrap();
-            while let Ok(Some(_)) = returned.next().await {
-                counter += 1
-            }
-            std::fs::remove_file(dbname).unwrap();
-            assert!(counter == 0)
-        })
-        .await;
+        let db = init_db(Some(&dbname)).await.unwrap();
+        let mut conn = db.connect().unwrap();
+        for file in filenames {
+            let path = PathBuf::from(file);
+            let tx = Transaction::new(&mut conn, Deferred).await.unwrap();
+            insert_file(tx, &path).await.unwrap();
+        }
+        let mut returned = conn.query(TOENCODE_PATHS, ()).await.unwrap();
+        while let Ok(Some(_)) = returned.next().await {
+            counter += 1
+        }
+        std::fs::remove_file(dbname).unwrap();
+        assert!(counter == 0)
     }
 
-    #[apply(test!)]
-    async fn check_update(ex: &Executor<'_>) {
+    #[tokio::test]
+    async fn check_update() {
         let dbname = PathBuf::from("temp2.db");
         let filenames = [
             "./samples/16bit.flac",
             "./samples/24bit.flac",
             "./samples/32bit.flac",
         ];
-        ex.spawn(async move {
-            let db = init_db(Some(&dbname)).await.unwrap();
-            let conn = db.connect().unwrap();
-            for file in filenames {
-                insert_file(&conn, &Path::new(file).canonicalize().unwrap())
-                    .await
-                    .unwrap();
-            }
-            conn.execute(
-                UPDATE_FILE,
-                params![
-                    Path::new("./samples/16bit.flac")
-                        .canonicalize()
-                        .unwrap()
-                        .to_str()
-                        .unwrap(),
-                    true,
-                    ""
-                ],
-            )
-            .await
-            .unwrap();
+        let db = init_db(Some(&dbname)).await.unwrap();
+        let mut conn = db.connect().unwrap();
+        for file in filenames {
+            let tx = Transaction::new(&mut conn, Deferred).await.unwrap();
+            insert_file(tx, &Path::new(file).canonicalize().unwrap())
+                .await
+                .unwrap();
+        }
+        conn.execute(
+            UPDATE_FILE,
+            params![
+                Path::new("./samples/16bit.flac")
+                    .canonicalize()
+                    .unwrap()
+                    .to_str()
+                    .unwrap(),
+                true,
+                ""
+            ],
+        )
+        .await
+        .unwrap();
 
-            update_file(
-                &conn,
-                &Path::new("./samples/16bit.flac").canonicalize().unwrap(),
-            )
-            .await
-            .unwrap();
+        let tx = Transaction::new(&mut conn, Deferred).await.unwrap();
 
-            let mut returned = conn.query(TOENCODE_PATHS, ()).await.unwrap();
-            let mut counter = 0;
-            while let Ok(Some(_)) = returned.next().await {
-                counter += 1
-            }
-            std::fs::remove_file(dbname).unwrap();
-            assert!(counter == 0)
-        })
-        .await;
+        update_file(
+            tx,
+            &Path::new("./samples/16bit.flac").canonicalize().unwrap(),
+        )
+        .await
+        .unwrap();
+
+        let mut returned = conn.query(TOENCODE_PATHS, ()).await.unwrap();
+        let mut counter = 0;
+        while let Ok(Some(_)) = returned.next().await {
+            counter += 1
+        }
+        std::fs::remove_file(dbname).unwrap();
+        assert!(counter == 0)
     }
 }
diff --git a/src/files.rs b/src/files.rs
index 78947a4..f27f78f 100644
--- a/src/files.rs
+++ b/src/files.rs
@@ -15,7 +15,11 @@ use std::{
     thread::{self, sleep},
     time::{Duration, UNIX_EPOCH},
 };
-use turso::Connection;
+use tokio::fs;
+use turso::{
+    Connection, Database,
+    transaction::{DropBehavior, Transaction, TransactionBehavior},
+};
 use walkdir::WalkDir;
 
 #[cfg(not(test))]
@@ -51,28 +55,27 @@ impl Display for FileError {
 
 impl Error for FileError {}
 
-async fn handle_file(file: &Path, conn: &Connection) -> Result<()> {
-    if db::check_file(conn, file).await? {
-        let modtime = file
-            .metadata()?
+async fn handle_file<'a>(file: &Path, tx: Transaction<'a>) -> Result<()> {
+    if db::check_file(&tx, &file).await? {
+        let modtime = fs::metadata(&file)
+            .await?
             .modified()?
             .duration_since(UNIX_EPOCH)?
             .as_secs();
-        let db_modtime = db::get_modtime(conn, file).await?;
+        let db_modtime = db::get_modtime(&tx, &file).await?;
         if modtime != db_modtime {
-            db::update_file(conn, file).await?;
+            db::update_file(tx, &file).await?;
         }
-        return Ok(());
+    } else {
+        db::insert_file(tx, &file).await?;
     }
 
-    db::insert_file(conn, file).await?;
-
     Ok(())
 }
 
-pub async fn index_files_recursively(
+pub async fn index_files_recursively<'a>(
     path: &Path,
-    conn: &Connection,
+    db: &Database,
     handler: Arc<AtomicBool>,
 ) -> Result<()> {
     if !path.is_dir() {
@@ -84,47 +87,48 @@ pub async fn index_files_recursively(
     let bar = ProgressBar::with_draw_target(Some(0), ProgressDrawTarget::stdout_with_hz(60))
         .with_style(ProgressStyle::with_template(BAR_TEMPLATE)?.progress_chars("#>-"))
         .with_message("Indexing");
-    let (filesend, filerecv) = mpsc::channel();
 
-    #[cfg(not(test))]
-    let newbar = bar.clone();
-
-    let newhandler = handler.clone();
+    let mut tasks = tokio::task::JoinSet::new();
 
     #[allow(unused_variables)]
-    thread::spawn(move || {
-        for entry in WalkDir::new(&abspath) {
-            if newhandler.load(Ordering::SeqCst) {
-                if let Err(error) = entry {
-                    #[cfg(not(test))]
-                    newbar.println(format!("{}", error));
-                } else {
-                    let path = entry.unwrap().into_path();
-                    if !path.is_file() {
-                        continue;
-                    }
-                    if path.extension().is_some_and(|x| x == "flac") {
-                        let _ = filesend.send(path.to_owned());
+    for entry in WalkDir::new(&abspath) {
+        if let Err(error) = entry {
+            #[cfg(not(test))]
+            bar.println(format!("{}", error));
+        } else {
+            let path = entry.unwrap().into_path();
+            if !path.is_file() {
+                continue;
+            }
+            if path.extension().is_some_and(|x| x == "flac") {
+                let mut conn = db.connect()?;
+
+                #[cfg(not(test))]
+                let newbar = bar.clone();
+                tasks.spawn(async move {
+                    let tx = Transaction::new(&mut conn, TransactionBehavior::Deferred)
+                        .await
+                        .unwrap();
+                    if let Err(error) = handle_file(&path, tx).await {
                         #[cfg(not(test))]
-                        newbar.inc_length(1);
+                        newbar.println(format!("{}", FileError::new(&path, error)));
                     } else {
-                        break;
+                        #[cfg(not(test))]
+                        newbar.inc(1);
                     }
-                }
+                });
+                #[cfg(not(test))]
+                bar.inc_length(1);
+            } else {
+                break;
             }
         }
-    });
+    }
 
-    while let Ok(path) = filerecv.recv()
-        && handler.load(Ordering::SeqCst)
-    {
-        #[allow(unused_variables)]
-        if let Err(error) = smol::block_on(async { handle_file(&path, conn).await }) {
-            #[cfg(not(test))]
-            bar.println(format!("{}", FileError::new(&path, error)));
-        } else {
-            #[cfg(not(test))]
-            bar.inc(1);
+    while let Some(_) = tasks.join_next().await {
+        if !handler.load(Ordering::SeqCst) {
+            tasks.shutdown();
+            break;
         }
     }
 
@@ -139,24 +143,36 @@ pub async fn index_files_recursively(
     Ok(())
 }
 
-pub async fn reencode_files(
+pub fn reencode_files(
     conn: &Connection,
     handler: Arc<AtomicBool>,
     threads: usize,
+    runtime: tokio::runtime::Runtime
 ) -> Result<()> {
+
+    let file_vec = runtime.block_on(async {db::get_toencode_files(conn).await})?;
+
     #[cfg(not(test))]
     let bar = ProgressBar::with_draw_target(
-        Some(db::get_toencode_number(conn).await?),
+        Some(file_vec.len() as u64),
         ProgressDrawTarget::stdout_with_hz(60),
     )
     .with_style(ProgressStyle::with_template(BAR_TEMPLATE)?.progress_chars("#>-"))
     .with_message("Reencoding");
+    let thread_counter = Arc::new(AtomicUsize::new(0));
 
-    let mut files = db::get_toencode_files(conn).await?.into_iter();
+    let (tx, rx) = std::sync::mpsc::channel();
 
-    let thread_counter = Arc::new(AtomicUsize::new(0));
+    let files = file_vec.iter();
 
     thread::scope(|s| {
+        s.spawn(move || {
+            let rt = tokio::runtime::Runtime::new().unwrap();
+
+            while let Ok(file) = rx.recv() {
+                
+            }
+        });
         while handler.load(Ordering::SeqCst) {
             if thread_counter.load(Ordering::Relaxed) >= threads {
                 sleep(Duration::from_millis(100));
@@ -176,14 +192,14 @@ pub async fn reencode_files(
             #[cfg(not(test))]
             let bar = bar.clone();
             let thread_counter = thread_counter.clone();
-            let conn = conn.clone();
 
             s.spawn(move || {
                 match handle_encode(&file, handler) {
                     Err(error) => eprintln!("{}", FileError::new(&file, error)),
                     Ok(false) => {
+
                         if let Err(error) =
-                            smol::block_on(async { db::update_file(&conn, &file).await })
+                            tokio::(async { db::update_file(&conn, &file).await })
                         {
                             eprintln!("{}", FileError::new(&file, error))
                         }
@@ -209,7 +225,7 @@ pub async fn reencode_files(
 }
 
 pub async fn clean_files(conn: &Connection, handler: Arc<AtomicBool>) -> Result<()> {
-    let files = db::init_clean_files(conn).await?;
+    let files = db::fetch_files(conn).await?;
 
     #[cfg(not(test))]
     let spinner = ProgressBar::with_draw_target(None, ProgressDrawTarget::stdout_with_hz(60))
@@ -259,6 +275,7 @@ mod tests {
         .await
     }
 
+    #[should_panic]
     #[apply(test!)]
     async fn test_clean_files(ex: &Executor<'_>) {
         let dbname = PathBuf::from("temp4.db");
@@ -281,7 +298,7 @@ mod tests {
             std::fs::remove_file("./samples/nonexisting.flac").unwrap();
 
             clean_files(&conn, handler).await.unwrap();
-            let counter = db::init_clean_files(&conn).await.unwrap().len();
+            let counter = db::fetch_files(&conn).await.unwrap().len();
             std::fs::remove_file(dbname).unwrap();
             assert!(counter == 3)
         })
diff --git a/src/main.rs b/src/main.rs
index d38d0d3..cb5ff66 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -89,33 +89,32 @@ fn main() -> Result<()> {
     ctrlc::set_handler(move || {
         r.store(false, Ordering::SeqCst);
     })?;
+    let runtime = tokio::runtime::Builder::new_multi_thread().build()?;
 
-    smol::block_on(async move {
-        let path = args.get_one::<PathBuf>("db");
-        let db = db::init_db(path).await?;
+    let path = args.get_one::<PathBuf>("db");
+    let db = runtime.block_on(async { db::init_db(path).await })?;
 
-        if path.is_none() && !args.get_flag("clean") && !args.get_flag("doit") {
-            let count = db::get_toencode_number(&db.connect()?).await?;
-            println!("Files to reencode:\t{}", style(count).green());
-            return Ok(());
-        }
+    if path.is_none() && !args.get_flag("clean") && !args.get_flag("doit") {
+        let count = runtime.block_on(async { db::get_toencode_number(&db.connect()?).await })?;
+        println!("Files to reencode:\t{}", style(count).green());
+        return Ok(());
+    }
 
-        if let Some(realpath) = path {
-            let hanlder = running.clone();
-            files::index_files_recursively(realpath, &db.connect()?, hanlder).await?;
-        }
+    if let Some(realpath) = path {
+        let hanlder = running.clone();
+        runtime.block_on(async { files::index_files_recursively(realpath, &db, hanlder).await })?;
+    }
 
-        if args.get_flag("clean") {
-            let handler = running.clone();
-            files::clean_files(&db.connect()?, handler).await?;
-        }
+    if args.get_flag("clean") {
+        let handler = running.clone();
+        runtime.block_on(async { files::clean_files(&db.connect()?, handler).await })?;
+    }
 
-        if args.get_flag("doit") {
-            let hanlder = running.clone();
-            let threads = *args.get_one::<usize>("threads").unwrap();
-            files::reencode_files(&db.connect()?, hanlder, threads).await?;
-        }
+    if args.get_flag("doit") {
+        let hanlder = running.clone();
+        let threads = *args.get_one::<usize>("threads").unwrap();
+        files::reencode_files(&db, hanlder, threads, runtime);
+    }
 
-        Ok::<(), anyhow::Error>(())
-    })
+    Ok::<(), anyhow::Error>(())
 }