testing reencoding

3 files changed, 68 insertions, 50 deletions

diff --git a/src/db.rs b/src/db.rs
index ea8d5d2..fba81c5 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -1,11 +1,10 @@
 use anyhow::{Result, anyhow};
 use directories::BaseDirs;
-use libsql::{Builder, Connection, params};
-use smol::stream::Stream;
 use std::{
     path::Path,
     time::{Duration, UNIX_EPOCH},
 };
+use turso::{Builder, Connection, Rows, params};
 
 use crate::flac::{CURRENT_VENDOR, get_vendor};
 
@@ -26,7 +25,10 @@ pub struct Database(Connection);
 
 impl Database {
     pub async fn new(path: impl AsRef<Path>) -> Result<Self> {
-        let conn = Builder::new_local(path).build().await?.connect()?;
+        let conn = Builder::new_local(path.as_ref().to_str().unwrap())
+            .build()
+            .await?
+            .connect()?;
         conn.execute(TABLE_CREATE, ()).await?;
 
         Ok(Database(conn))
@@ -78,7 +80,7 @@ impl Database {
             .next()
             .await?
         {
-            Ok(matches!(row.get_value(0)?, libsql::Value::Integer(1)))
+            Ok(matches!(row.get_value(0)?, turso::Value::Integer(1)))
         } else {
             Err(anyhow!("database error"))
         }
@@ -102,11 +104,9 @@ impl Database {
         }
     }
 
-    pub async fn init_clean_files(
-        &self,
-    ) -> Result<impl Stream<Item = libsql::Result<libsql::Row>>> {
+    pub async fn init_clean_files(&self) -> Result<Rows, turso::Error> {
         self.0.execute(DEDUPE_DB, ()).await?;
-        Ok(self.0.query(FETCH_FILES, ()).await?.into_stream())
+        self.0.query(FETCH_FILES, ()).await
     }
 
     pub async fn remove_file(&self, filename: impl AsRef<Path>) -> Result<()> {
@@ -116,21 +116,21 @@ impl Database {
         Ok(())
     }
 
-    pub async fn get_toencode_stream(
-        &self,
-    ) -> Result<impl Stream<Item = libsql::Result<libsql::Row>>> {
-        Ok(self.0.query(TOENCODE_QUERY, ()).await?.into_stream())
+    pub async fn get_toencode_files(&self) -> Result<Rows, turso::Error> {
+        self.0.query(TOENCODE_QUERY, ()).await
     }
 
-    pub async fn get_toencode_number(&self) -> Result<u64> {
-        Ok(self
+    pub async fn get_toencode_number(&self) -> Result<i64> {
+        Ok(*self
             .0
             .query(TOENCODE_NUMBER, ())
             .await?
             .next()
             .await?
             .unwrap()
-            .get::<u64>(0)?)
+            .get_value(0)?
+            .as_integer()
+            .unwrap())
     }
 
     pub async fn vaccum(&self) -> Result<()> {
@@ -150,7 +150,6 @@ pub async fn open_default_db() -> Result<Database> {
 
 #[cfg(test)]
 mod tests {
-    use futures_util::StreamExt;
     use macro_rules_attribute::apply;
     use smol_macros::{Executor, test};
 
@@ -166,15 +165,9 @@ mod tests {
             for file in filenames {
                 let _ = conn.insert_file(&file.to_string()).await;
             }
-            let returned = conn
-                .0
-                .query(TOENCODE_QUERY, ())
-                .await
-                .unwrap()
-                .into_stream();
-            pin_utils::pin_mut!(returned);
+            let mut returned = conn.0.query(TOENCODE_QUERY, ()).await.unwrap();
 
-            while let Some(Ok(_)) = returned.next().await {
+            while let Ok(Some(_)) = returned.next().await {
                 counter += 1
             }
             std::fs::remove_file(dbname).unwrap();
@@ -221,15 +214,9 @@ mod tests {
             .await
             .unwrap();
 
-            let returned = conn
-                .0
-                .query(TOENCODE_QUERY, ())
-                .await
-                .unwrap()
-                .into_stream();
-            pin_utils::pin_mut!(returned);
+            let mut returned = conn.0.query(TOENCODE_QUERY, ()).await.unwrap();
             let mut counter = 0;
-            while let Some(Ok(_)) = returned.next().await {
+            while let Ok(Some(_)) = returned.next().await {
                 counter += 1
             }
             std::fs::remove_file(dbname).unwrap();
diff --git a/src/files.rs b/src/files.rs
index 52789ce..13ba06a 100644
--- a/src/files.rs
+++ b/src/files.rs
@@ -134,39 +134,71 @@ pub fn index_files_recursively(
                 #[cfg(not(test))]
                 bar.inc_length(1);
             }
+        } else {
+            break;
         }
     }
 
     tasks.par_iter_mut().for_each(|task| {
-        if let Err(error) = smol::block_on(async { ex.run(task).await }) {
-            eprintln!("{error}")
+        if running.load(Ordering::SeqCst) {
+            if let Err(error) = smol::block_on(async { ex.run(task).await }) {
+                eprintln!("{error}")
+            }
         }
     });
 
     #[cfg(not(test))]
     {
-        if !running.load(Ordering::SeqCst) {
-            bar.abandon_with_message("Indexing aborted");
-        } else {
+        if running.load(Ordering::SeqCst) {
             bar.finish_with_message("Finished indexing");
+        } else {
+            bar.abandon_with_message("Indexing aborted");
         }
     }
     Ok(())
 }
 
-/* pub fn reencode_files(conn: &Database) -> Result<()> {
-    let stream = conn.get_toencode_stream().await?;
-    pin_mut!(stream);
+async fn get_reencode_vec(conn: &Database) -> Result<Vec<PathBuf>> {
+    let mut files = Vec::new();
+    let mut rows = conn.get_toencode_files().await?;
+    while let Ok(Some(row)) = rows.next().await {
+        files.push(PathBuf::from(row.get_value(0)?.as_text().unwrap()))
+    }
+    Ok(files)
+}
 
+pub fn reencode_files(conn: &Database, running: Arc<AtomicBool>) -> Result<()> {
     #[cfg(not(test))]
     let bar = ProgressBar::with_draw_target(
-        Some(conn.get_toencode_number().await?),
+        Some(smol::block_on(async { conn.get_toencode_number().await })?.try_into()?),
         ProgressDrawTarget::stdout_with_hz(60),
     )
     .with_style(ProgressStyle::with_template(BAR_TEMPLATE)?.progress_chars("#>-"))
     .with_message("Reencoding");
 
-    while let Some(Ok(row)) = stream.next().await {
+    let files = smol::block_on(async { get_reencode_vec(conn).await })?;
+
+    files.par_iter().for_each(|file| {
+        if running.load(Ordering::SeqCst) {
+            std::thread::sleep(Duration::from_secs(3));
+            #[cfg(not(test))]
+            bar.inc(1);
+        } else {
+            eprintln!("{}", FileError::new(file, anyhow!("cancelled")))
+        }
+    });
+
+    #[cfg(not(test))]
+    {
+        if running.load(Ordering::SeqCst) {
+            bar.finish_with_message("Finished reencoding");
+        } else {
+            bar.abandon_with_message("Reencoding aborted");
+        }
+    }
+    Ok(())
+
+    /* while let Some(Ok(row)) = stream.next().await {
         let filename = Path::new(row.get_str(0)?).canonicalize()?;
         if filename.exists() {
             let newconn = conn.clone();
@@ -197,12 +229,10 @@ pub fn index_files_recursively(
                 }
             });
         }
-    }
-
-    Ok(())
+    } */
 }
 
-pub fn clean_files(conn: &Database) -> Result<()> {
+/* pub fn clean_files(conn: &Database) -> Result<()> {
     let ex = Executor::new();
 
     let mut tasks: JoinSet<std::result::Result<(), anyhow::Error>> = JoinSet::new();
diff --git a/src/main.rs b/src/main.rs
index 6db57a3..0e28b1f 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -114,12 +114,13 @@ fn main() -> Result<()> {
         pool.install(|| files::index_files_recursively(realpath, &conn, hanlder))?;
     }
 
-    /*     if args.get_flag("clean") {
+    /* if args.get_flag("clean") {
         pool.install(|| files::clean_files(&conn))?;
-    }
+    } */
 
     if args.get_flag("doit") {
-        files::reencode_files(&conn)?;
-    } */
+        let hanlder = running.clone();
+        pool.install(|| files::reencode_files(&conn, hanlder))?;
+    }
     Ok::<(), anyhow::Error>(())
 }