Merge pull request #325 from originalworks/owen-output-generator-refactor

cezary-stroczynski · web-flow · commit 3bf8971fd1fc · 2025-10-17T14:23:00.000+02:00
Owen output generator refactor
diff --git a/owen/src/aws-lambda/main.rs b/owen/src/aws-lambda/main.rs
@@ -88,11 +88,11 @@ async fn function_handler(
     println!("synced directories: {s3_message_folders:?}");
 
     match owen::run_with_sentry(&owen_config).await {
-        Ok(message_processing_context) => {
+        Ok(ddex_messages) => {
             queue
                 .sync_message_folder_statuses(
                     local_to_s3_folder_mapping,
-                    message_processing_context,
+                    ddex_messages,
                     s3_message_folders,
                 )
                 .await
diff --git a/owen/src/aws-lambda/message_queue.rs b/owen/src/aws-lambda/message_queue.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
 use aws_sdk_dynamodb::types::AttributeValue;
-use owen::output_generator::MessageDirProcessingContext;
+use owen::output_generator::DdexMessage;
 use std::{collections::HashMap, env};
 
 pub struct MessageQueue {
@@ -156,29 +156,27 @@ impl MessageQueue {
     pub async fn sync_message_folder_statuses(
         &self,
         local_to_s3_folder_mapping: HashMap<String, String>,
-        message_processing_context_vec: Vec<MessageDirProcessingContext>,
+        ddex_messages: Vec<DdexMessage>,
         message_folders: Vec<String>,
     ) -> Result<()> {
-        let mut s3_folder_to_processing_context_map: HashMap<String, MessageDirProcessingContext> =
-            HashMap::new();
+        let mut s3_folder_to_ddex_message_map: HashMap<String, DdexMessage> = HashMap::new();
 
-        for message_processing_context in message_processing_context_vec {
+        for ddex_message in ddex_messages {
             let s3_path = local_to_s3_folder_mapping
-                .get(&message_processing_context.message_dir_path)
+                .get(&ddex_message.message_dir_path)
                 .expect(
                     format!(
                         "Could not retrieve s3 path from mapping to local folder. Local folder: {}",
-                        message_processing_context.message_dir_path
+                        ddex_message.message_dir_path
                     )
                     .as_str(),
                 );
 
-            s3_folder_to_processing_context_map.insert(s3_path.clone(), message_processing_context);
+            s3_folder_to_ddex_message_map.insert(s3_path.clone(), ddex_message);
         }
         for folder in message_folders {
-            let message_processing_context = s3_folder_to_processing_context_map.get(&folder);
-            if let Some(message_processing_context) = message_processing_context {
-                if message_processing_context.excluded {
+            if let Some(ddex_message) = s3_folder_to_ddex_message_map.get(&folder) {
+                if ddex_message.excluded {
                     self.set_single_message_folder_status(
                         folder.clone(),
                         self.rejected_status_value.to_string(),
diff --git a/owen/src/lib.rs b/owen/src/lib.rs
@@ -15,12 +15,13 @@ use contracts::ContractsManager;
 use ddex_parser::ParserError;
 pub use log;
 use log_macros::{format_error, log_error};
-use output_generator::MessageDirProcessingContext;
 use sentry::User;
 use serde_json::json;
 use std::env;
 use std::str::FromStr;
 
+use crate::ipfs::IpfsManager;
+use crate::output_generator::{DdexMessage, OutputFilesGenerator};
 use crate::wallet::OwenWallet;
 
 #[cfg(any(feature = "aws-integration", feature = "local-s3"))]
@@ -45,7 +46,7 @@ pub enum IpfsInterface {
 pub struct Config {
     pub rpc_url: String,
     pub private_key: Option<String>,
-    pub folder_path: String,
+    pub input_files_dir: String,
     pub local_ipfs: bool,
     pub output_files_dir: String,
     pub username: String,
@@ -75,7 +76,7 @@ impl Config {
         let mut args = std::env::args();
         args.next();
 
-        let folder_path = args
+        let input_files_dir = args
             .next()
             .unwrap_or_else(|| Config::get_env_var("INPUT_FILES_DIR").to_string());
 
@@ -137,7 +138,7 @@ impl Config {
         let config = Config {
             rpc_url,
             private_key,
-            folder_path,
+            input_files_dir,
             local_ipfs,
             ipfs_api_base_url,
             output_files_dir,
@@ -179,13 +180,14 @@ impl Config {
     }
 }
 
-pub async fn run(config: &Config) -> anyhow::Result<Vec<MessageDirProcessingContext>> {
+pub async fn run(config: &Config) -> anyhow::Result<Vec<DdexMessage>> {
     let owen_wallet = OwenWallet::build(&config).await?;
     let contracts_manager = ContractsManager::build(&config, &owen_wallet).await?;
     contracts_manager.check_image_compatibility().await?;
 
-    let message_dir_processing_log =
-        output_generator::create_output_files(&config, &owen_wallet).await?;
+    let ipfs_manager = IpfsManager::build(&config, &owen_wallet).await?;
+    let output_files_generator = OutputFilesGenerator::build(&config, &ipfs_manager)?;
+    let ddex_messages = output_files_generator.generate_files().await?;
 
     let blob_transaction_data = BlobTransactionData::build(&config.output_files_dir)?;
 
@@ -205,19 +207,18 @@ pub async fn run(config: &Config) -> anyhow::Result<Vec<MessageDirProcessingCont
     } else {
         contracts_manager.send_blob(blob_transaction_data).await?;
     }
-
-    Ok(message_dir_processing_log)
+    Ok(ddex_messages)
 }
 
-pub async fn run_with_sentry(config: &Config) -> anyhow::Result<Vec<MessageDirProcessingContext>> {
+pub async fn run_with_sentry(config: &Config) -> anyhow::Result<Vec<DdexMessage>> {
     sentry::configure_scope(|scope| {
         scope.set_user(Some(User {
             username: Some(config.username.to_owned()),
             ..Default::default()
         }));
     });
 
-    let message_dir_processing_context = run(&config).await.map_err(|e| {
+    let ddex_messages = run(&config).await.map_err(|e| {
         sentry::configure_scope(|scope| {
             scope.set_tag("error_type", {
                 if e.is::<ParserError>() {
@@ -232,5 +233,5 @@ pub async fn run_with_sentry(config: &Config) -> anyhow::Result<Vec<MessageDirPr
         log_error!("{e}")
     })?;
 
-    anyhow::Ok(message_dir_processing_context)
+    anyhow::Ok(ddex_messages)
 }
diff --git a/owen/src/local-s3/main.rs b/owen/src/local-s3/main.rs
@@ -34,15 +34,15 @@ async fn main() -> Result<()> {
     let s3_message_folders = storage.s3_message_folders.clone();
 
     match owen::run_with_sentry(&owen_config).await {
-        Ok(message_processing_context) => {
-            let s3_folder_to_processing_context_map = database.save_message_folders(
+        Ok(ddex_messages) => {
+            let s3_folder_to_ddex_message_map = database.save_message_folders(
                 local_to_s3_folder_mapping,
-                message_processing_context,
+                ddex_messages,
                 &s3_message_folders,
             )?;
 
             storage
-                .clear_s3_folders(s3_folder_to_processing_context_map, &s3_message_folders)
+                .clear_s3_folders(s3_folder_to_ddex_message_map, &s3_message_folders)
                 .await?;
         }
         Err(e)
diff --git a/owen/src/local-s3/message_database.rs b/owen/src/local-s3/message_database.rs
@@ -1,7 +1,7 @@
 use anyhow::Result;
 use owen::{
     constants::{DEFAULT_DATABASE_NAME, DEFAULT_TABLE_NAME},
-    output_generator::MessageDirProcessingContext,
+    output_generator::DdexMessage,
 };
 use rusqlite::{params, Connection};
 use std::collections::HashMap;
@@ -48,34 +48,28 @@ impl MessageDatabase {
     pub fn save_message_folders(
         &self,
         local_to_s3_folder_mapping: HashMap<String, String>,
-        message_processing_context_vec: Vec<MessageDirProcessingContext>,
+        ddex_messages: Vec<DdexMessage>,
         message_folders: &Vec<String>,
-    ) -> Result<HashMap<String, MessageDirProcessingContext>> {
-        let mut s3_folder_to_processing_context_map: HashMap<String, MessageDirProcessingContext> =
-            HashMap::new();
+    ) -> Result<HashMap<String, DdexMessage>> {
+        let mut s3_folder_to_ddex_message_map: HashMap<String, DdexMessage> = HashMap::new();
 
-        for message_processing_context in message_processing_context_vec {
+        for ddex_message in ddex_messages {
             let s3_path = local_to_s3_folder_mapping
-                .get(&message_processing_context.message_dir_path)
+                .get(&ddex_message.message_dir_path)
                 .expect(
                     format!(
                         "Could not retrieve s3 path from mapping to local folder. Local folder: {}",
-                        message_processing_context.message_dir_path
+                        ddex_message.message_dir_path
                     )
                     .as_str(),
                 );
 
-            s3_folder_to_processing_context_map.insert(s3_path.clone(), message_processing_context);
+            s3_folder_to_ddex_message_map.insert(s3_path.clone(), ddex_message);
         }
         for s3_message_folder in message_folders {
-            let message_processing_context =
-                s3_folder_to_processing_context_map.get(s3_message_folder);
-            if let Some(message_processing_context) = message_processing_context {
-                if &message_processing_context.excluded == &true {
-                    let reason = message_processing_context
-                        .reason
-                        .as_ref()
-                        .expect("Could not retrieve exclusion reason from processing context");
+            if let Some(ddex_message) = s3_folder_to_ddex_message_map.get(s3_message_folder) {
+                if &ddex_message.excluded == &true {
+                    let reason = ddex_message.reason.as_deref().unwrap_or("");
                     self.connection.execute(
                         format!(
                             "INSERT INTO {} (message_folder, status, reason) VALUES (?1, ?2, ?3)",
@@ -105,7 +99,7 @@ impl MessageDatabase {
                 )?;
             }
         }
-        Ok(s3_folder_to_processing_context_map)
+        Ok(s3_folder_to_ddex_message_map)
     }
 
     pub fn save_message_folders_with_status(
diff --git a/owen/src/output_generator.rs b/owen/src/output_generator.rs
diff --git a/owen/src/s3_message_storage.rs b/owen/src/s3_message_storage.rs
