started implementing async logic

3 files changed, 68 insertions, 54 deletions

diff --git a/src/db.rs b/src/db.rs
index e75d86f..420fdb7 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -19,7 +19,7 @@ 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";
 
-async fn init_db(path: Option<&Path>) -> Result<turso::Database> {
+pub(crate) async fn init_db(path: Option<&Path>) -> Result<turso::Database> {
     let db = if let Some(file) = path {
         turso::Builder::new_local(file.to_str().unwrap())
             .build()
@@ -37,7 +37,7 @@ async fn init_db(path: Option<&Path>) -> Result<turso::Database> {
     Ok(db)
 }
 
-async fn insert_file(conn: &Connection, filename: &Path) -> Result<()> {
+pub(crate) async fn insert_file(conn: &Connection, filename: &Path) -> Result<()> {
     let toencode = !matches!(get_vendor(filename)?.as_str(), CURRENT_VENDOR);
 
     let modtime = filename
@@ -55,7 +55,7 @@ async fn insert_file(conn: &Connection, filename: &Path) -> Result<()> {
     Ok(())
 }
 
-async fn update_file(conn: &Connection, filename: &Path) -> Result<()> {
+pub(crate) async fn update_file(conn: &Connection, filename: &Path) -> Result<()> {
     let modtime = filename
         .metadata()?
         .modified()?
@@ -71,7 +71,7 @@ async fn update_file(conn: &Connection, filename: &Path) -> Result<()> {
     Ok(())
 }
 
-async fn check_file(conn: &Connection, filename: &Path) -> Result<bool> {
+pub(crate) async fn check_file(conn: &Connection, filename: &Path) -> Result<bool> {
     Ok(conn
         .query(CHECK_FILE, params!(filename.to_str().unwrap()))
         .await?
@@ -81,28 +81,36 @@ async fn check_file(conn: &Connection, filename: &Path) -> Result<bool> {
         .get::<bool>(0)?)
 }
 
-async fn init_clean_files(conn: &Connection) -> Result<Vec<PathBuf>, turso::Error> {
+pub(crate) async fn init_clean_files(conn: &Connection) -> Result<Vec<PathBuf>, turso::Error> {
     conn.execute(DEDUPE_DB, ()).await?;
     let mut rows = conn.query(FETCH_FILES, ()).await?;
     let mut files = Vec::new();
     while let Ok(Some(row)) = rows.next().await {
-        let path = row.get::<String>(0)?;
+        let path: String = row.get(0)?;
         files.push(PathBuf::from(path));
+    }
+
+    Ok(files)
+}
 
-async fn remove_file(conn: &Connection, filename: &Path) -> Result<()> {
+pub(crate) async fn remove_file(conn: &Connection, filename: &Path) -> Result<()> {
     conn.execute(REMOVE_FILE, params!(filename.to_str().unwrap()))
         .await?;
     Ok(())
 }
 
-async fn get_toencode_files(conn: &Connection) -> Result<Vec<PathBuf>, turso::Error> {
+pub(crate) async fn get_toencode_files(conn: &Connection) -> Result<Vec<PathBuf>, turso::Error> {
     let mut rows = conn.query(TOENCODE_PATHS, ()).await?;
     let mut files: Vec<PathBuf> = Vec::new();
     while let Ok(Some(row)) = rows.next().await {
-        let path = row.get::<String>(0)?;
+        let path: String = row.get(0)?;
         files.push(PathBuf::from(path));
+    }
 
-async fn get_toencode_number(conn: &Connection) -> Result<u64, turso::Error> {
+    Ok(files)
+}
+
+pub(crate) async fn get_toencode_number(conn: &Connection) -> Result<u64, turso::Error> {
     Ok(conn
         .query(TOENCODE_NUMBER, ())
         .await?
@@ -112,7 +120,7 @@ async fn get_toencode_number(conn: &Connection) -> Result<u64, turso::Error> {
         .get::<u64>(0)?)
 }
 
-async fn get_modtime(conn: &Connection, file: &Path) -> Result<u64> {
+pub(crate) async fn get_modtime(conn: &Connection, file: &Path) -> Result<u64> {
     Ok(conn
         .query(GET_MODTIME, params![file.to_str().unwrap()])
         .await?
@@ -122,12 +130,11 @@ async fn get_modtime(conn: &Connection, file: &Path) -> Result<u64> {
         .get::<u64>(0)?)
 }
 
-async fn vacuum(conn: &Connection) -> Result<()> {
+pub(crate) async fn vacuum(conn: &Connection) -> Result<()> {
     conn.execute("VACUUM", ()).await?;
     Ok(())
 }
 
-
 #[cfg(test)]
 mod tests {
 
diff --git a/src/files.rs b/src/files.rs
index afb42d7..71310c4 100644
--- a/src/files.rs
+++ b/src/files.rs
@@ -1,7 +1,8 @@
+use crate::db;
+use crate::flac::handle_encode;
 use anyhow::{Result, anyhow};
 #[cfg(not(test))]
 use indicatif::{ProgressBar, ProgressDrawTarget, ProgressStyle};
-use rusqlite::Connection;
 use std::{
     error::Error,
     fmt::Display,
@@ -13,10 +14,9 @@ use std::{
     thread::{self, sleep},
     time::{Duration, UNIX_EPOCH},
 };
+use turso::Connection;
 use walkdir::WalkDir;
 
-use crate::{db::Database, flac::handle_encode};
-
 #[cfg(not(test))]
 const BAR_TEMPLATE: &str = "{msg:<} [{wide_bar:.green/cyan}] Elapsed: {elapsed} {pos:>7}/{len:7}";
 #[cfg(not(test))]
@@ -50,21 +50,21 @@ impl Display for FileError {
 
 impl Error for FileError {}
 
-fn handle_file(file: &Path, conn: &Connection) -> Result<()> {
-    if conn.check_file(file)? {
+async fn handle_file(file: &Path, conn: &Connection) -> Result<()> {
+    if db::check_file(conn, file).await? {
         let modtime = file
             .metadata()?
             .modified()?
             .duration_since(UNIX_EPOCH)?
             .as_secs();
-        let db_modtime = conn.get_modtime(file)?;
+        let db_modtime = db::get_modtime(conn, file).await?;
         if modtime != db_modtime {
-            conn.update_file(file)?;
+            db::update_file(conn, file).await?;
         }
         return Ok(());
     }
 
-    conn.insert_file(file)?;
+    db::insert_file(conn, file).await?;
 
     Ok(())
 }
@@ -106,7 +106,7 @@ pub fn index_files_recursively(
                 continue;
             }
             if path.extension().is_some_and(|x| x == "flac") {
-                if let Err(error) = handle_file(&path, conn) {
+                if let Err(error) = smol::block_on(async { handle_file(&path, conn).await }) {
                     eprintln!("{}", FileError::new(&path, error));
                 } else {
                     #[cfg(not(test))]
@@ -129,16 +129,20 @@ pub fn index_files_recursively(
     Ok(())
 }
 
-pub fn reencode_files(conn: Connection, handler: Arc<AtomicBool>, threads: usize) -> Result<()> {
+pub async fn reencode_files(
+    conn: Connection,
+    handler: Arc<AtomicBool>,
+    threads: usize,
+) -> Result<()> {
     #[cfg(not(test))]
     let bar = ProgressBar::with_draw_target(
-        Some(conn.get_toencode_number()?),
+        Some(db::get_toencode_number(&conn).await?),
         ProgressDrawTarget::stdout_with_hz(60),
     )
     .with_style(ProgressStyle::with_template(BAR_TEMPLATE)?.progress_chars("#>-"))
     .with_message("Reencoding");
 
-    let mut files = conn.get_toencode_files()?.into_iter();
+    let mut files = db::get_toencode_files(&conn).await?.into_iter();
 
     let lock = Arc::new(Mutex::new(conn));
 
@@ -166,11 +170,11 @@ pub fn reencode_files(conn: Connection, handler: Arc<AtomicBool>, threads: usize
             let bar = bar.clone();
             let thread_counter = thread_counter.clone();
 
-            s.spawn(move || {
+            s.spawn(async move || {
                 match handle_encode(&file, handler) {
                     Err(error) => eprintln!("{}", FileError::new(&file, error)),
                     Ok(false) => {
-                        if let Err(error) = lock.lock().unwrap().update_file(&file) {
+                        if let Err(error) = db::update_file(&lock.lock().unwrap(), &file).await {
                             eprintln!("{}", FileError::new(&file, error));
                         }
                         #[cfg(not(test))]
diff --git a/src/main.rs b/src/main.rs
index d1e4562..2ef2491 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,11 +1,10 @@
-mod db;
-mod files;
-mod flac;
+pub(crate) mod db;
+pub(crate) mod files;
+pub(crate) mod flac;
 use anyhow::Result;
 use clap::{Arg, ArgAction, Command, ValueHint, command, value_parser};
 use clap_complete::{Generator, Shell, generate};
 use console::style;
-use rusqlite::Connection;
 use std::{
     path::PathBuf,
     sync::{
@@ -14,8 +13,6 @@ use std::{
     },
 };
 
-use crate::db::Database;
-
 fn build_cli() -> Command {
     command!()
         .arg(
@@ -93,30 +90,36 @@ fn main() -> Result<()> {
         r.store(false, Ordering::SeqCst);
     })?;
 
-    let conn = Connection::new(args.get_one::<PathBuf>("db"))?;
+    smol::block_on(async move {
+        let path = if let Some(path) = args.get_one::<PathBuf>("db") {
+            Some(path.as_path())
+        } else {
+            None
+        };
+        let db = db::init_db(path).await?;
 
-    let path = args.get_one::<PathBuf>("path");
+        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 = conn.get_toencode_number()?;
-        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)?;
+        }
 
-    if let Some(realpath) = path {
-        let hanlder = running.clone();
-        files::index_files_recursively(realpath, &conn, hanlder)?;
-    }
+        if args.get_flag("clean") {
+            let handler = running.clone();
+            files::clean_files(&db.connect()?, handler)?;
+        }
 
-    if args.get_flag("clean") {
-        let handler = running.clone();
-        files::clean_files(&conn, handler)?;
-    }
+        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(conn, hanlder, threads)?;
-    }
-    Ok::<(), anyhow::Error>(())
+        Ok::<(), anyhow::Error>(())
+    })
 }