From 1fea9b682b6ed08b17495374a8ede21b51101faa Mon Sep 17 00:00:00 2001
From: jakka <jakkadoujin@gmail.com>
Date: Thu, 19 Jun 2025 14:39:55 +0300
Subject: reworked encoding async logic

---
 src/files.rs | 59 +++++++++++++++++++++++++++++++++++++++++------------------
 src/flac.rs  | 12 +-----------
 src/main.rs  |  1 -
 3 files changed, 42 insertions(+), 30 deletions(-)

(limited to 'src')

diff --git a/src/files.rs b/src/files.rs
index a019cfb..6f2bce2 100644
--- a/src/files.rs
+++ b/src/files.rs
@@ -2,13 +2,15 @@ use anyhow::{Result, anyhow};
 use futures_util::StreamExt;
 use pin_utils::pin_mut;
 use std::{
+    error::Error,
     fmt::Display,
     path::{Path, PathBuf},
     time::UNIX_EPOCH,
 };
 use tokio::{fs::read_dir, task::JoinSet};
+use walkdir::WalkDir;
 
-use crate::{db::Database, flac::handle_encode};
+use crate::{db::Database, flac::encode_file};
 
 #[derive(Debug)]
 pub struct FileError {
@@ -36,6 +38,8 @@ impl Display for FileError {
     }
 }
 
+impl Error for FileError {}
+
 async fn handle_file(file: impl AsRef<Path>, conn: Database) -> Result<()> {
     match conn.check_file(&file).await {
         Ok(true) => {
@@ -64,11 +68,30 @@ async fn handle_file(file: impl AsRef<Path>, conn: Database) -> Result<()> {
     Ok(())
 }
 
+async fn count_flacs(path: impl AsRef<Path>) -> u64 {
+    let mut counter = 0;
+    let _ = WalkDir::new(path)
+        .into_iter()
+        .map(|file| {
+            let path = file.unwrap().into_path();
+            if path.is_file() && path.extension().unwrap() == "flac" {
+                counter += 1
+            }
+        })
+        .collect::<Vec<_>>();
+    counter
+}
+
 pub async fn index_files_recursively(path: impl AsRef<Path>, conn: &Database) -> Result<()> {
     if !path.as_ref().is_dir() {
         return Err(anyhow!("Invalid root directory"));
     }
     let abspath = path.as_ref().canonicalize()?;
+
+    let files = count_flacs(&abspath).await;
+
+    println!("Total flacs:\t{files}");
+
     let mut tasks = JoinSet::new();
 
     let mut dirs = vec![abspath];
@@ -117,30 +140,30 @@ pub async fn reencode_files(conn: &Database) -> Result<()> {
     while let Some(Ok(row)) = stream.next().await {
         if let Some(file) = row.get_value(0)?.as_text() {
             let filename = Path::new(file).canonicalize()?;
-            tasks.spawn_blocking(move || handle_encode(filename));
+            let newconn = conn.clone();
+            tasks.spawn(async move {
+                let file = filename.clone();
+                if let Err(error) = tokio::task::spawn_blocking(move || encode_file(file)).await? {
+                    return Err(anyhow!(FileError::new(&filename, error)));
+                };
+
+                if let Err(error) = newconn.update_file(&filename).await {
+                    return Err(anyhow!(FileError::new(&filename, error)));
+                };
+
+                Ok(())
+            });
         }
     }
 
-    let mut update_tasks = JoinSet::new();
-
     while let Some(task) = tasks.join_next().await {
         match task {
-            Ok(Ok(path)) => {
-                let newconn = conn.clone();
-                update_tasks.spawn(async move { newconn.update_file(path).await });
+            Ok(Err(error)) => eprintln!("Error encountered:\t{error}"),
+            Err(error) => eprintln!("Error encountered:\t{error}"),
+            _ => {
                 counter += 1;
-                print!("\rReencoded files:\t{counter}")
+                print!("Reencoded:\t{counter}");
             }
-            Ok(Err(error)) => eprintln!("{error}"),
-            Err(error) => eprintln!("Error encountered:\t{}", error),
-        }
-    }
-
-    while let Some(task) = update_tasks.join_next().await {
-        match task {
-            Ok(Err(error)) => eprintln!("{error}"),
-            Err(error) => eprintln!("Error encountered:\t{}", error),
-            _ => {}
         }
     }
 
diff --git a/src/flac.rs b/src/flac.rs
index 42f9d21..97f61b5 100644
--- a/src/flac.rs
+++ b/src/flac.rs
@@ -15,8 +15,6 @@ use symphonia::core::{
     meta::MetadataOptions,
 };
 
-use crate::files;
-
 pub const CURRENT_VENDOR: &str = "reference libFLAC 1.5.0 20250211";
 
 type BoxedFormatReader = Box<dyn FormatReader>;
@@ -281,7 +279,7 @@ fn encode_cycle_32(
     Ok(hasher.finalize().to_vec())
 }
 
-fn encode_file(filename: impl AsRef<Path>) -> Result<()> {
+pub fn encode_file(filename: impl AsRef<Path>) -> Result<()> {
     let filencoder = FileEncoder::new(filename)?;
 
     let mut outf = File::create(filencoder.temp_name())?;
@@ -299,14 +297,6 @@ fn encode_file(filename: impl AsRef<Path>) -> Result<()> {
     filencoder.encode(enc)
 }
 
-pub fn handle_encode(file: impl AsRef<Path>) -> Result<impl AsRef<Path>> {
-    if let Err(error) = encode_file(&file) {
-        Err(anyhow!(files::FileError::new(file, error)))
-    } else {
-        Ok(file)
-    }
-}
-
 pub fn get_vendor(file: impl AsRef<Path>) -> Result<String> {
     if let Some(vorbis) = Tag::read_from_path(file)?.vorbis_comments() {
         Ok(vorbis.vendor_string.to_owned())
diff --git a/src/main.rs b/src/main.rs
index da04a9e..bb2a716 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -19,7 +19,6 @@ fn build_cli() -> Command {
             Arg::new("doit")
                 .long("doit")
                 .help("Actually reencode files")
-                .conflicts_with("index")
                 .action(ArgAction::SetTrue),
         )
         .arg(
-- 
cgit v1.3.1