added more file processing logic

5 files changed, 94 insertions, 33 deletions

diff --git a/Cargo.lock b/Cargo.lock
index ffd4d64..4df4ada 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -845,9 +845,9 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
 
 [[package]]
 name = "libc"
-version = "0.2.172"
+version = "0.2.173"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d750af042f7ef4f724306de029d18836c26c1765a54a6a3f094cbd23a7267ffa"
+checksum = "d8cfeafaffdbc32176b64fb251369d52ea9f0a8fbc6f8759edffef7b525d64bb"
 
 [[package]]
 name = "libflac-sys"
@@ -1383,6 +1383,7 @@ dependencies = [
  "libsql",
  "md-5",
  "metaflac",
+ "pin-utils",
  "symphonia",
  "tokio",
 ]
diff --git a/Cargo.toml b/Cargo.toml
index 1a544e7..769ed5e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -14,6 +14,7 @@ i24 = "2.1.0"
 libsql = { version = "0.9.10", default-features = false, features = ["core", "sync"] }
 md-5 = "0.10.6"
 metaflac = "0.2.8"
+pin-utils = "0.1.0"
 symphonia = { version = "0.5.4", path = "../Symphonia/symphonia", default-features = false, features = ["flac"] }
 tokio = { version = "1.45.1", features = ["macros", "rt", "rt-multi-thread"] }
 #symphonia = { git = "https://github.com/pdeljanov/Symphonia.git", branch = "dev-0.6", default-features = false, features = ["flac"] }
diff --git a/src/db.rs b/src/db.rs
index b261e2c..1016894 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -1,7 +1,8 @@
 use anyhow::{Result, anyhow};
 use directories::BaseDirs;
-use futures_util::StreamExt;
+use futures_util::Stream;
 use libsql::{Builder, Connection, params};
+use pin_utils::pin_mut;
 use std::{
     ffi::OsStr,
     fmt::Display,
@@ -83,21 +84,6 @@ impl Database {
         Ok(())
     }
 
-    pub async fn get_files_toencode(&self) -> Result<Vec<String>> {
-        let rows = self.0.query(TOENCODE_QUERY, ()).await?;
-        if rows.column_count() == 0 {
-            return Err(anyhow!(Errors::EmptyQuery));
-        };
-
-        let filenames = rows
-            .into_stream()
-            .map(|row| row.unwrap().get_str(0).unwrap().to_string())
-            .collect::<Vec<String>>()
-            .await;
-
-        Ok(filenames)
-    }
-
     pub async fn check_file(&self, filename: &impl AsRef<OsStr>) -> Result<bool> {
         let abs_filename = absolute(Path::new(filename))?;
 
@@ -152,6 +138,12 @@ 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 open_default_db() -> Result<Database> {
@@ -165,6 +157,8 @@ pub async fn open_default_db() -> Result<Database> {
 
 #[cfg(test)]
 mod tests {
+    use futures_util::StreamExt;
+
     use super::*;
 
     #[tokio::test]
@@ -175,9 +169,22 @@ mod tests {
         for file in filenames {
             let _ = conn.insert_file(&file.to_string()).await;
         }
-        let returned = conn.get_files_toencode().await.unwrap();
+        let returned = conn
+            .0
+            .query(TOENCODE_QUERY, ())
+            .await
+            .unwrap()
+            .into_stream();
+        pin_mut!(returned);
+
+        let mut counter = 0;
+
+        while let Some(Ok(_)) = returned.next().await {
+            counter += 1
+        }
+
         std::fs::remove_file(dbname).unwrap();
-        assert!(returned.is_empty())
+        assert!(counter == 0)
     }
 
     #[tokio::test]
@@ -203,8 +210,18 @@ mod tests {
 
         conn.update_file(&"16bit.flac".to_string()).await.unwrap();
 
-        let returned = conn.get_files_toencode().await.unwrap();
+        let returned = conn
+            .0
+            .query(TOENCODE_QUERY, ())
+            .await
+            .unwrap()
+            .into_stream();
+        pin_mut!(returned);
+        let mut counter = 0;
+        while let Some(Ok(_)) = returned.next().await {
+            counter += 1
+        }
         std::fs::remove_file(dbname).unwrap();
-        assert!(returned.is_empty())
+        assert!(counter == 0)
     }
 }
diff --git a/src/files.rs b/src/files.rs
index e0d9757..dc96650 100644
--- a/src/files.rs
+++ b/src/files.rs
@@ -1,5 +1,6 @@
 use anyhow::{Result, anyhow};
-use libsql::Connection;
+use futures_util::StreamExt;
+use pin_utils::pin_mut;
 use std::{
     fmt::Display,
     path::{Path, PathBuf, absolute},
@@ -7,14 +8,20 @@ use std::{
 };
 use tokio::{fs::read_dir, task::JoinSet};
 
-use crate::db::Database;
+use crate::{db::Database, flac::handle_encode};
 
 #[derive(Debug)]
-struct FileError {
+pub struct FileError {
     file: PathBuf,
     error: anyhow::Error,
 }
 
+impl FileError {
+    pub fn new(file: PathBuf, error: anyhow::Error) -> Self {
+        FileError { file, error }
+    }
+}
+
 impl Display for FileError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
@@ -37,17 +44,17 @@ async fn handle_file(file: PathBuf, conn: Database) -> Result<()> {
             let db_time = conn.get_modtime(&file).await?;
             if modtime != db_time {
                 if let Err(error) = conn.update_file(&file).await {
-                    return Err(anyhow!(FileError { file, error }));
+                    return Err(anyhow!(FileError::new(file, error)));
                 };
             }
             return Ok(());
         }
-        Err(error) => return Err(anyhow!(FileError { file, error })),
+        Err(error) => return Err(anyhow!(FileError::new(file, error))),
         _ => {}
     }
 
     if let Err(error) = conn.insert_file(&file).await {
-        return Err(anyhow!(FileError { file, error }));
+        return Err(anyhow!(FileError::new(file, error)));
     }
 
     Ok(())
@@ -59,7 +66,6 @@ pub async fn index_files_recursively(path: &Path, conn: &Database) -> Result<()>
     }
     let abspath = absolute(path)?;
     let mut tasks = JoinSet::new();
-    let mut counter: i64 = 0;
 
     let mut dirs = vec![abspath];
 
@@ -74,12 +80,34 @@ pub async fn index_files_recursively(path: &Path, conn: &Database) -> Result<()>
                 if let Some(ext) = path.extension() {
                     if ext == "flac" {
                         let newconn = conn.clone();
-                        counter += 1;
                         tasks.spawn(async move { handle_file(path, newconn).await });
                     }
                 }
             }
-            print!("\rFiles found:\t{counter}")
+        }
+    }
+
+    while let Some(task) = tasks.join_next().await {
+        match task {
+            Ok(Err(error)) => eprintln!("{error}"),
+            Err(error) => eprintln!("Error encountered:\t{}", error),
+            _ => {}
+        }
+    }
+
+    Ok(())
+}
+
+pub async fn reencode_files(conn: &Database) -> Result<()> {
+    let stream = conn.get_toencode_stream().await?;
+    pin_mut!(stream);
+
+    let mut tasks = JoinSet::new();
+
+    while let Some(Ok(row)) = stream.next().await {
+        if let Some(file) = row.get_value(0)?.as_text() {
+            let filename = PathBuf::from(file.clone());
+            tasks.spawn_blocking(move || handle_encode(filename));
         }
     }
 
diff --git a/src/flac.rs b/src/flac.rs
index 16627e2..940e816 100644
--- a/src/flac.rs
+++ b/src/flac.rs
@@ -3,7 +3,11 @@ use flac_bound::{FlacEncoder, WriteWrapper};
 use i24::i24;
 use md5::{Digest, Md5, Md5Core, digest::core_api::CoreWrapper};
 use metaflac::{Block, Tag};
-use std::{ffi::OsStr, fs::File, path::Path};
+use std::{
+    ffi::OsStr,
+    fs::File,
+    path::{Path, PathBuf},
+};
 use symphonia::core::{
     audio::{Audio, GenericAudioBufferRef},
     codecs::audio::AudioDecoder,
@@ -12,6 +16,8 @@ use symphonia::core::{
     meta::MetadataOptions,
 };
 
+use crate::files;
+
 pub const CURRENT_VENDOR: &str = "reference libFLAC 1.5.0 20250211";
 
 struct StreamConfig {
@@ -246,7 +252,7 @@ fn write_tags(
     Ok(())
 }
 
-pub fn encode_file(filename: impl AsRef<OsStr>) -> Result<()> {
+fn encode_file(filename: impl AsRef<OsStr>) -> Result<()> {
     let file = Path::new(&filename);
     let tempname = &format!("{}.tmp", file.to_str().unwrap());
 
@@ -279,6 +285,14 @@ pub fn encode_file(filename: impl AsRef<OsStr>) -> Result<()> {
     Ok(())
 }
 
+pub fn handle_encode(file: PathBuf) -> Result<()> {
+    if let Err(error) = encode_file(&file) {
+        Err(anyhow!(files::FileError::new(file, error)))
+    } else {
+        Ok(())
+    }
+}
+
 pub fn get_vendor(file: &Path) -> Result<String> {
     let tag = Tag::read_from_path(file)?;
     Ok(tag.vorbis_comments().unwrap().vendor_string.clone())