feat: recipe repo update routine

- wip: firmware service
- wip: remote shell

Signed-off-by: Pakin <pakin.t@forth.co.th>

src/app.rs

@@ -18,7 +18,10 @@ use reqwest::{StatusCode, multipart};
 use std::{
     collections::{HashMap, VecDeque},
     env,
-    sync::{Arc, RwLock},
+    sync::{
+        Arc, RwLock,
+        atomic::{AtomicI64, AtomicU32, AtomicU64},
+    },
     time::Duration,
 };
 use tokio::sync::{Mutex, mpsc::Sender};
@@ -35,6 +38,7 @@ pub struct DevConfig {
     pub api_domain: String,
     pub api_recipe_service: String,
     pub api_redis_url: String,
+    pub api_git_taobin_project: String,
     pub api_sheet_endpoints: Arc<Mutex<Vec<String>>>,
     pub shared_configures: Arc<RwLock<serde_json::Value>>,
     pub allowed_origins: Vec<String>,
@@ -46,6 +50,7 @@ impl DevConfig {
         domain: String,
         rp_service: String,
         api_redis_url: String,
+        api_git_taobin_project: String,
         api_sheet_endpoints: Arc<Mutex<Vec<String>>>,
         shared_configures: Arc<RwLock<serde_json::Value>>,
     ) -> DevConfig {
@@ -55,6 +60,7 @@ impl DevConfig {
             api_recipe_service: rp_service,
             api_redis_url,
             api_sheet_endpoints,
+            api_git_taobin_project,
             shared_configures,
             allowed_origins: Vec::new(),
         }
@@ -69,10 +75,18 @@ impl DevConfig {
         format!("{}{}", self.api_domain, self.api_recipe_service)
     }
 
+    pub fn get_taobin_project_url(&self) -> String {
+        format!("{}", self.api_git_taobin_project)
+    }
+
     pub fn get_file_from_recipe_repo(&self, path: String) -> String {
         format!("{}/checkout?path={}", self.get_recipe_url(), path)
     }
 
+    pub fn get_file_from_taobin_project_repo(&self, path: String) -> String {
+        format!("{}/checkout?path={}", self.get_taobin_project_url(), path)
+    }
+
     pub fn get_post_file_to_recipe_repo(&self) -> String {
         format!("{}/commit", self.get_recipe_url())
     }
@@ -85,6 +99,10 @@ impl DevConfig {
         format!("{}/push", self.get_recipe_url())
     }
 
+    pub fn get_replay_recipe_repo(&self) -> String {
+        format!("{}/replay", self.get_recipe_url())
+    }
+
     pub fn get_api_header(&self) -> (String, String) {
         ("X-API-Key".to_string(), self.api_key.clone())
     }
@@ -211,6 +229,7 @@ impl AppState {
     ) -> Arc<AppState> {
         let redis_cli_clone = redis_cli.clone();
         let tx_new = system_tx.clone();
+        let tx2 = system_tx.clone();
 
         let mut interceptor = crate::websocket::interceptor::create_interceptor_client(&dev_config);
         if let Some(ref mut ic) = interceptor {
@@ -253,13 +272,112 @@ impl AppState {
                 clients: HashMap::new(),
             })),
             interceptor,
-            http_client,
+            http_client: http_client.clone(),
             debug,
             firebase_project_id,
             jwk_encoding_keys: RwLock::new(Vec::new()),
         });
         // NOTE: removed backup process, let each app handled by themselves
 
+        let client_clone = http_client.clone();
+
+        // Background task for actively sync with recipe repo
+        let self_clone2 = result.clone();
+        tokio::spawn(async move {
+            loop {
+                info!("[RecipeRoutine] Syncing recipe repo.");
+                if let Err(e) =
+                    invoke_pull_sync_recipe_request(&client_clone, dev_config.clone()).await
+                {
+                    warn!("[RecipeRoutine][Sync] Retry again in another 30 seconds. Error: {e}");
+                    tokio::time::sleep(Duration::from_secs(30)).await;
+                    continue;
+                }
+
+                let commits: Vec<String> = {
+                    let pcli = self_clone2.postgres_cli.lock().await;
+                    match pcli
+                        .query(
+                            "SELECT commit_hash FROM recovery_commit WHERE repo = 'recipe_repo'",
+                            &[],
+                        )
+                        .await
+                    {
+                        Ok(q) => q
+                            .iter()
+                            .map(|x| {
+                                let r: String = x.get("commit_hash");
+
+                                r
+                            })
+                            .collect(),
+                        Err(_) => Vec::new(),
+                    }
+                };
+
+                if !commits.is_empty() {
+                    // wait replay
+                    if let Err(e) = invoke_replay_recipe_repo(
+                        &client_clone,
+                        dev_config.clone(),
+                        commits.clone(),
+                    )
+                    .await
+                    {
+                        warn!(
+                            "[RecipeRoutine][Replay] Retry again in another 30 seconds. Error: {e}"
+                        );
+                        tokio::time::sleep(Duration::from_secs(30)).await;
+                        continue;
+                    }
+
+                    info!("[RecipeRoutine] Has {} commits, pushing ...", commits.len());
+                    if let Err(e) =
+                        invoke_push_recipe_repo_request(&client_clone, dev_config.clone()).await
+                    {
+                        warn!(
+                            "[RecipeRoutine][Push] Retry again in another 30 seconds. Error: {e}"
+                        );
+                        tokio::time::sleep(Duration::from_secs(30)).await;
+                        continue;
+                    }
+
+                    // success push, do update postgres
+
+                    {
+                        let mut pcli = self_clone2.postgres_cli.lock().await;
+                        if let Ok(ptx) = pcli.transaction().await {
+                            match ptx
+                                .execute(
+                                    "DELETE FROM recovery_commit WHERE commit_hash = ANY($1)",
+                                    &[&commits],
+                                )
+                                .await
+                            {
+                                Ok(_) => match ptx.commit().await {
+                                    Ok(_) => {
+                                        info!("[RecipeRoutine][UpdateTx] Successfully updated.");
+                                    }
+                                    Err(e) => {
+                                        error!(
+                                            "[RecipeRoutine][UpdateTx] Failed to commit tx: {e}"
+                                        );
+                                    }
+                                },
+                                Err(e) => {
+                                    error!(
+                                        "[RecipeRoutine][UpdateTx] Failed to remove pending commits: {e}"
+                                    );
+                                }
+                            }
+                        }
+                    }
+                }
+
+                tokio::time::sleep(Duration::from_secs(300)).await;
+            }
+        });
+
         // Background task for refresh Google's keys daily
         let self_clone = result.clone();
         tokio::spawn(async move {
@@ -267,10 +385,21 @@ impl AppState {
                 if let Err(e) = refresh_jwk_cache(Arc::clone(&self_clone)).await {
                     error!("Failed tp updating background JWKS keys: {e:?}");
                 }
+
+                let _ = tx2.send(serde_json::json!({
+                  "type": "notify-ks",
+                  "payload": {
+                    "from": "system_tx",
+                    "to": "*",
+                    "ks-type": "KeyUpdate"
+                  }
+                }));
+
                 tokio::time::sleep(Duration::from_secs(86400)).await;
             }
         });
 
+        // Command Background Listener Task
         tokio::spawn(async move {
             let mut lredis = redis_cli_clone.clone();
             let current_queue: crossbeam_queue::ArrayQueue<CommandRequestPayload> =
@@ -538,7 +667,7 @@ async fn pprof_block() -> impl axum::response::IntoResponse {
     )
 }
 
-pub async fn invoke_checkout_request(
+pub async fn invoke_recipe_checkout_request(
     http_client: &reqwest::Client,
     config: DevConfig,
     path: String,
@@ -552,11 +681,25 @@ pub async fn invoke_checkout_request(
     }
 }
 
-/// Invoke git pull, may takes sometime
+pub async fn invoke_taobin_project_checkout_request(
-pub async fn invoke_pull_sync_request(
     http_client: &reqwest::Client,
     config: DevConfig,
+    path: String,
 ) -> Result<String, Box<dyn std::error::Error>> {
+    let req_path = config.get_file_from_taobin_project_repo(path);
+    let res = http_client.get(req_path).send().await?;
+
+    match res.text().await {
+        Ok(raw) => Ok(raw),
+        Err(e) => Err(format!("{e}").into()),
+    }
+}
+
+/// Invoke git pull, may takes sometime
+async fn invoke_pull_sync_recipe_request(
+    http_client: &reqwest::Client,
+    config: DevConfig,
+) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
     let req_path = config.get_pull_recipe_repo();
     let res = http_client.get(req_path).send().await?;
 
@@ -575,6 +718,48 @@ pub async fn invoke_pull_sync_request(
     }
 }
 
+async fn invoke_push_recipe_repo_request(
+    http_client: &reqwest::Client,
+    config: DevConfig,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let req_path = config.get_push_recipe_repo();
+    let res = http_client.get(req_path).send().await?;
+
+    if res.status() != StatusCode::OK {
+        error!(
+            "invoke push fail: [{}] {:?}",
+            res.status(),
+            res.text().await
+        );
+        return Err("push fail".into());
+    }
+
+    Ok(())
+}
+
+async fn invoke_replay_recipe_repo(
+    http_client: &reqwest::Client,
+    config: DevConfig,
+    commit_hashes: Vec<String>,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let response = http_client
+        .post(config.get_replay_recipe_repo())
+        .body(
+            serde_json::json!({
+              "commit_hashes": commit_hashes
+            })
+            .to_string(),
+        )
+        .send()
+        .await;
+
+    if response.is_err() {
+        return Err(response.unwrap_err().into());
+    }
+
+    Ok(())
+}
+
 /// Invoke sending from server to server for committing
 pub async fn invoke_commit_request(
     http_client: &reqwest::Client,
@@ -727,6 +912,9 @@ pub async fn initialize() -> Result<(), Box<dyn std::error::Error>> {
     let api_redis = env::var("DEV_API_REDIS").unwrap_or("0.0.0.0".to_string());
     let api_redis_port = env::var("DEV_API_REDIS_PORT").unwrap_or("6379".to_string());
 
+    let api_git_taobin_project =
+        env::var("DEV_TBJ_GIT_SERVICE").expect("no provided taobin project service");
+
     // No need for resolver
     // let api_resolver = env::var("RESOLVER_SERVICE_URL").expect("no available resolver");
 
@@ -747,15 +935,17 @@ pub async fn initialize() -> Result<(), Box<dyn std::error::Error>> {
         api_domain,
         api_recipe_service,
         format!("redis://{api_redis}:{api_redis_port}"),
+        api_git_taobin_project,
         Arc::new(Mutex::new(sheet_endpoint_config)),
         Arc::new(RwLock::new(shared_configures)),
     );
     dev_cfg = dev_cfg.with_allowed_origins(&allowed_origins).clone();
 
+    info!("redis: {}", dev_cfg.api_redis_url.clone());
+
     let redis_cli = redis::Client::open(dev_cfg.api_redis_url.clone())?;
 
-    let (mut client, connection) =
+    let (client, connection) = tokio_postgres::connect(&postgres_connection_config, NoTls).await?;
-        tokio_postgres::connect(&postgres_connection_config, NoTls).await?;
     tokio::spawn(async move {
         if let Err(e) = connection.await {
             error!("connection postgres error: {e}");

src/websocket/handler.rs

@@ -370,10 +370,13 @@ async fn handle_socket(
     let auth_result = timeout(Duration::from_secs(2), async {
         if let Some(Ok(Message::Text(text))) = receiver.next().await {
             let handshake: HandshakePayload = serde_json::from_str(&text)?;
-            info!("handshake ok!");
+            info!(
+                "[{temp_session}] handshake with version {:?}",
+                handshake.client_version
+            );
             // Offline JWT validation using memory cache
             let uid = verify_token(&handshake.token, state_clone).await?;
-            info!("uid: {uid}");
+            info!("[{temp_session}] uid: {uid}");
             // Execute Ephemeral Elliptic Curve DF Key Exchange
             let (server_pub_b64, cipher) = execute_dh_handshake(&handshake.client_public_key)?;
 
@@ -383,7 +386,7 @@ async fn handle_socket(
                 server_public_key: server_pub_b64,
             })?;
 
-            // info!("ack sending ... {ack_payload}");
+            info!("[{temp_session}] ack sending ...");
 
             sender.send(Message::Text(ack_payload.into())).await?;
 
@@ -404,7 +407,9 @@ async fn handle_socket(
     let session = match auth_result {
         Ok(Ok(valid_session)) => Arc::new(valid_session),
         _ => {
-            warn!("Connection dropped: Handshake timeout or authentication failed");
+            warn!(
+                "[{temp_session}] Connection dropped: Handshake timeout or authentication failed"
+            );
             let _ = sender
                 .send(Message::Close(Some(axum::extract::ws::CloseFrame {
                     code: 4001,
@@ -425,7 +430,7 @@ async fn handle_socket(
             let mut ulock = user.lock().await;
             *ulock = valid_uid.clone();
         }
-        info!("update user uid");
+        info!("[{temp_session}] update user uid to {}", session.uid);
     }
 
     {

src/websocket/model.rs

@@ -90,17 +90,6 @@ pub struct CommandRequestPayload {
     pub values: serde_json::Value,
 }
 
-/// For logging user's action
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct LogReportPayload {
-    // expect either `email` or `unknown`
-    pub user: String,
-    pub action: String,
-    // expect either country name or `unknown dep`
-    pub country: String,
-    pub values: serde_json::Value,
-}
-
 /// Message for saving recipe
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct SaveRecipePayload {
@@ -307,3 +296,66 @@ impl CreateMaterial {
                 ));
     }
 }
+
+/// Remote shell request from server to client, for executing adb command on user's connected Android device.
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RemoteShellPayload {
+    /// Unique ID for this request (used to match the response).
+    pub generated_request_id: String,
+
+    /// The ADB shell command to execute on the device.
+    pub command_input: String,
+
+    /// Human-readable explanation of why this command should run.
+    pub purpose_declaration: String,
+
+    /// If true, the client MUST show a confirmation dialog
+    /// before executing. If absent/false, the command runs immediately
+    /// (unless flagged as dangerous — the client then forces confirmation).
+    #[serde(default)]
+    pub require_confirmation: bool,
+
+    /// Optional timeout in seconds. When set, the client auto-aborts
+    /// the command after this duration and returns partial output.
+    pub timeout: Option<u32>,
+
+    /// Target uid of user
+    pub to: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RemoteShellResponsePayload {
+    /// The original request_id from the server.
+    pub request_id: String,
+
+    /// true if exitCode === 0, false otherwise.
+    pub success: bool,
+
+    /// stdout captured from the command.
+    pub output: String,
+
+    /// stderr captured from the command.
+    pub error: String,
+
+    /// Process exit code. -1 if timed out or no device connected.
+    pub exit_code: i32,
+
+    /// Echo back the user_info from the request.
+    pub user_info: serde_json::Value,
+
+    /// Unix millisecond timestamp of completion.
+    pub timestamp: u64,
+
+    /// Present and true when the command was aborted by timeout.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub timed_out: Option<bool>,
+}
+
+/// Request to start finalize for building firmware.
+#[derive(Debug, Serialize, Deserialize)]
+pub struct FinalizeFirmwareRequest {
+    pub country: String,
+    pub requester_info: serde_json::Value,
+}

src/websocket/rw.rs

@@ -180,21 +180,24 @@ pub async fn read(
                                         continue;
                                     }
                                 }
-                                "log_report" if let Some(log_payload) = req.payload => {
+                                // NOTE: discontinued as it is unused and not planned to use
-                                    let log_report_payload: LogReportPayload =
+                                //
-                                        match safe_deserialize(&log_payload) {
+                                //
-                                            Ok(lreq) => lreq,
+                                // "log_report" if let Some(log_payload) = req.payload => {
-                                            Err(e) => {
+                                //     let log_report_payload: LogReportPayload =
-                                                error!(
+                                //         match safe_deserialize(&log_payload) {
-                                                    "error deserialize body log request: {e:?} ---> Skip"
+                                //             Ok(lreq) => lreq,
-                                                );
+                                //             Err(e) => {
-                                                continue;
+                                //                 error!(
-                                            }
+                                //                     "error deserialize body log request: {e:?} ---> Skip"
-                                        };
+                                //                 );
-                                    // generate timestamp
+                                //                 continue;
-                                    //
+                                //             }
-                                    let now = Instant::now();
+                                //         };
-                                }
+                                //     // generate timestamp
+                                //     //
+                                //     let now = Instant::now();
+                                // }
                                 "save_recipe" if req.payload.is_some() => {
                                     if tasks::recipe::handle_recipe_save_change_request(
                                         &http_client,
@@ -252,6 +255,14 @@ pub async fn read(
                                         continue;
                                     }
                                 }
+                                "upload-log" if req.payload.is_some() => {
+                                    // TODO
+                                }
+                                "remote-shell-response" if req.payload.is_some() => {
+                                    // TODO
+                                    //
+                                }
+                                "firmware-request" if req.payload.is_some() => {}
                                 _ => {
                                     let uidd = uid.lock().await.clone();
                                     // not implemented

src/websocket/session.rs

@@ -17,7 +17,8 @@ pub(crate) struct SecureSession {
 #[derive(serde::Deserialize)]
 pub(crate) struct HandshakePayload {
     pub token: String,
-    pub client_public_key: String, // BASE 64
+    pub client_public_key: String,      // BASE 64
+    pub client_version: Option<String>, // See the version of client, as 0.0.2 is start of using secured but not sending this field yet
 }
 
 #[derive(serde::Serialize)]

src/websocket/tasks/firmware.rs

@@ -0,0 +1,75 @@
+use std::sync::Arc;
+
+use log::info;
+use redis::TypedCommands;
+use serde::{Deserialize, Serialize};
+use tokio::sync::{Mutex, mpsc::Sender};
+
+use crate::{
+    app::{DevConfig, invoke_recipe_checkout_request},
+    websocket::{
+        core::{TxControlMessage, WebsocketMessageResult, safe_deserialize},
+        model::{FinalizeFirmwareRequest, WebsocketMessageRequest},
+    },
+};
+
+#[derive(Debug, Serialize, Deserialize)]
+struct PartialFirmwareBuildArg {
+    // full tha param
+    pub country: String,
+    pub files: Vec<String>,
+}
+
+pub async fn handle_request_firmware(
+    http_client: &reqwest::Client,
+    config: DevConfig,
+    mut redis: redis::Client,
+    tx: Sender<TxControlMessage>,
+    req: WebsocketMessageRequest,
+    uid_clone: Arc<Mutex<String>>,
+) -> WebsocketMessageResult {
+    let p = req.payload.unwrap();
+    let req_build_firmware: FinalizeFirmwareRequest = safe_deserialize(&p)?;
+
+    let country = req_build_firmware.country;
+
+    // channel for each country
+    let channel_build = format!("build/{country}");
+
+    info!(
+        "request building firmware from {}",
+        req_build_firmware.requester_info
+    );
+
+    let firmware_request_param = PartialFirmwareBuildArg {
+        country,
+        param: "".to_string(),
+    };
+    let _ = redis.publish(
+        channel_build,
+        serde_json::to_string(&firmware_request_param)?,
+    );
+
+    match invoke_recipe_checkout_request(
+        &http_client,
+        config,
+        format!("{country}/last_build_release_date"),
+    )
+    .await
+    {
+        Ok(last_build) => {
+            if let Ok(date) = chrono::DateTime::parse_from_str(&last_build, LAST_BUILD_DATE_FORMAT)
+            {
+            } else {
+            }
+        }
+        Err(e) => {}
+    }
+
+    // TODO
+    // - filter files by date from repo
+    // - grep file path by filtered
+    // NOTE: we must have index init data for every country, must set somewhere.
+
+    Ok(())
+}

src/websocket/tasks/mod.rs

@@ -1,5 +1,6 @@
 pub mod auth;
 pub mod command;
+// pub mod firmware;
 pub mod price;
 pub mod recipe;
 pub mod sheet;

src/websocket/tasks/price.rs

@@ -148,11 +148,16 @@ pub async fn handle_price_request(
 
     let price_action = price_param.action;
 
-    let price_content =
+    let price_content = match invoke_recipe_checkout_request(
-        match invoke_checkout_request(http_client, config.clone(), price_file_format.clone()).await {
+        http_client,
-            Ok(pc) => pc,
+        config.clone(),
-            Err(e) => return Err(format!("Cannot find price of expected country: {e:?}").into()),
+        price_file_format.clone(),
-        };
+    )
+    .await
+    {
+        Ok(pc) => pc,
+        Err(e) => return Err(format!("Cannot find price of expected country: {e:?}").into()),
+    };
 
     info!("price content len: {}", price_content.len());
 

src/websocket/tasks/recipe.rs

@@ -73,7 +73,7 @@ async fn get_latest_recipe_from_git(
 ) -> Result<Recipe, Box<dyn std::error::Error>> {
     let latest_key = format!("{country}/version");
     let latest_version =
-        match invoke_checkout_request(http_client, config.clone(), latest_key).await {
+        match invoke_recipe_checkout_request(http_client, config.clone(), latest_key).await {
             Ok(version) => version,
             Err(e) => {
                 println!("Error on checkout: {e}");
@@ -86,13 +86,14 @@ async fn get_latest_recipe_from_git(
     for i in init_key..6 {
         let r1_key = get_key_cache(country.to_string(), latest_version.clone(), false, i);
 
-        let content = match invoke_checkout_request(http_client, config.clone(), r1_key).await {
+        let content =
-            Ok(file_content) => file_content,
+            match invoke_recipe_checkout_request(http_client, config.clone(), r1_key).await {
-            Err(e) => {
+                Ok(file_content) => file_content,
-                println!("Error on checkout: {e}");
+                Err(e) => {
-                "".to_string()
+                    println!("Error on checkout: {e}");
-            }
+                    "".to_string()
-        };
+                }
+            };
         info!("[get-latest] content ready: {}", content.len());
         let recipe = serde_json::from_str::<Recipe>(&content);
 
@@ -116,7 +117,8 @@ async fn get_latest_recipe_saved_machine_from_git(
 ) -> Result<Recipe, Box<dyn std::error::Error>> {
     let latest_key = format!("{country}/coffeethai02_{country}_{boxid}_temp.json");
     let content =
-        match invoke_checkout_request(http_client, config.clone(), latest_key.clone()).await {
+        match invoke_recipe_checkout_request(http_client, config.clone(), latest_key.clone()).await
+        {
             Ok(content) => content,
             Err(e) => {
                 println!("Error on checkout: {e}");
@@ -324,7 +326,7 @@ pub async fn handle_recipe_request(
     if latest_version.is_empty() {
         // cannot get actual version, try get from git
         latest_version =
-            match invoke_checkout_request(http_client, config.clone(), latest_key).await {
+            match invoke_recipe_checkout_request(http_client, config.clone(), latest_key).await {
                 Ok(version) => version,
                 Err(e) => {
                     println!("Error on checkout: {e}");
@@ -453,7 +455,9 @@ pub async fn handle_recipe_request(
                 } else {
                     // retry get from git
                     let content =
-                        match invoke_checkout_request(http_client, config.clone(), r1_key).await {
+                        match invoke_recipe_checkout_request(http_client, config.clone(), r1_key)
+                            .await
+                        {
                             Ok(file_content) => file_content,
                             Err(e) => {
                                 println!("Error on checkout: {e}");
@@ -498,7 +502,7 @@ pub async fn handle_recipe_versions_list_request(
     let version_list = format!("{country}", country = recipe_param.country);
 
     let country_versions_str =
-        match invoke_checkout_request(http_client, config.clone(), version_list).await {
+        match invoke_recipe_checkout_request(http_client, config.clone(), version_list).await {
             Ok(vs) => vs,
             Err(e) => return Err(format!("Cannot find versions of expected country: {e:?}").into()),
         };
@@ -694,7 +698,7 @@ pub async fn handle_request_list_menu_recipe(
     let latest_key = format!("{country}/version", country = req_menu_list.country);
 
     let latest_version =
-        match invoke_checkout_request(http_client, config.clone(), latest_key).await {
+        match invoke_recipe_checkout_request(http_client, config.clone(), latest_key).await {
             Ok(version) => version,
             Err(e) => {
                 println!("Error on checkout: {e}");
@@ -729,7 +733,8 @@ pub async fn handle_request_list_menu_recipe(
         );
 
         let content =
-            match invoke_checkout_request(http_client, config.clone(), r1_key.clone()).await {
+            match invoke_recipe_checkout_request(http_client, config.clone(), r1_key.clone()).await
+            {
                 Ok(file_content) => file_content,
                 Err(e) => {
                     println!("Error on checkout: {e}");
@@ -848,7 +853,11 @@ async fn modify_material(
 
     if payload.alarm_id_when_offline.is_some() {
         set_clauses.push(format!("alarm_id_when_offline = ${}", param_index));
-        let val = payload.alarm_id_when_offline.as_ref().map(|v| v.as_i64().unwrap_or(0) as i32).unwrap_or(0);
+        let val = payload
+            .alarm_id_when_offline
+            .as_ref()
+            .map(|v| v.as_i64().unwrap_or(0) as i32)
+            .unwrap_or(0);
         param_values.push(Box::new(val));
         param_index += 1;
     }
@@ -869,7 +878,11 @@ async fn modify_material(
 
     if payload.drain_timer.is_some() {
         set_clauses.push(format!("drain_timer = ${}", param_index));
-        let val = payload.drain_timer.as_ref().map(|v| v.as_i64().unwrap_or(0) as i32).unwrap_or(0);
+        let val = payload
+            .drain_timer
+            .as_ref()
+            .map(|v| v.as_i64().unwrap_or(0) as i32)
+            .unwrap_or(0);
         param_values.push(Box::new(val));
         param_index += 1;
     }
@@ -894,14 +907,22 @@ async fn modify_material(
 
     if payload.low_to_offline.is_some() {
         set_clauses.push(format!("low_to_offline = ${}", param_index));
-        let val = payload.low_to_offline.as_ref().map(|v| v.as_i64().unwrap_or(0) as i32).unwrap_or(0);
+        let val = payload
+            .low_to_offline
+            .as_ref()
+            .map(|v| v.as_i64().unwrap_or(0) as i32)
+            .unwrap_or(0);
         param_values.push(Box::new(val));
         param_index += 1;
     }
 
     if payload.material_status.is_some() {
         set_clauses.push(format!("material_status = ${}", param_index));
-        let val = payload.material_status.as_ref().map(|v| v.as_i64().unwrap_or(0) as i32).unwrap_or(0);
+        let val = payload
+            .material_status
+            .as_ref()
+            .map(|v| v.as_i64().unwrap_or(0) as i32)
+            .unwrap_or(0);
         param_values.push(Box::new(val));
         param_index += 1;
     }
@@ -932,7 +953,11 @@ async fn modify_material(
 
     if payload.schedule_drain_type.is_some() {
         set_clauses.push(format!("schedule_drain_type = ${}", param_index));
-        let val = payload.schedule_drain_type.as_ref().map(|v| v.as_i64().unwrap_or(0) as i32).unwrap_or(0);
+        let val = payload
+            .schedule_drain_type
+            .as_ref()
+            .map(|v| v.as_i64().unwrap_or(0) as i32)
+            .unwrap_or(0);
         param_values.push(Box::new(val));
         param_index += 1;
     }
@@ -957,7 +982,11 @@ async fn modify_material(
 
     if payload.id_alternate.is_some() {
         set_clauses.push(format!("id_alternate = ${}", param_index));
-        let val = payload.id_alternate.as_ref().map(|v| v.as_i64().unwrap_or(0) as i32).unwrap_or(0);
+        let val = payload
+            .id_alternate
+            .as_ref()
+            .map(|v| v.as_i64().unwrap_or(0) as i32)
+            .unwrap_or(0);
         param_values.push(Box::new(val));
         param_index += 1;
     }
@@ -994,7 +1023,11 @@ async fn modify_material(
 
     if payload.pay_rettry_max_count.is_some() {
         set_clauses.push(format!("pay_retry_max_count = ${}", param_index));
-        let val = payload.pay_rettry_max_count.as_ref().map(|v| v.as_i64().unwrap_or(0) as i32).unwrap_or(0);
+        let val = payload
+            .pay_rettry_max_count
+            .as_ref()
+            .map(|v| v.as_i64().unwrap_or(0) as i32)
+            .unwrap_or(0);
         param_values.push(Box::new(val));
         param_index += 1;
     }
@@ -1029,7 +1062,8 @@ async fn modify_material(
 
     param_values.push(Box::new(payload.id));
 
-    let params: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = param_values.iter().map(|b| b.as_ref() as _).collect();
+    let params: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> =
+        param_values.iter().map(|b| b.as_ref() as _).collect();
 
     match client.execute(&sql, &params).await {
         Ok(rows) => Ok(rows),
@@ -1208,9 +1242,14 @@ pub async fn handle_request_material_action(
                     && let Ok(modify_payload) = serde_json::from_value::<ModifyMaterial>(d)
                 {
                     let client = postgres_cli.lock().await;
-                    let tx_result = match modify_material(&client, &modify_payload, &tx, &uidd).await {
+                    let tx_result = match modify_material(&client, &modify_payload, &tx, &uidd)
+                        .await
+                    {
                         Ok(rows_updated) => {
-                            info!("[modify_material] updated {} row(s) for id={}", rows_updated, modify_payload.id);
+                            info!(
+                                "[modify_material] updated {} row(s) for id={}",
+                                rows_updated, modify_payload.id
+                            );
                             serde_json::json!({
                                 "type": "notify",
                                 "payload": {
@@ -1222,7 +1261,10 @@ pub async fn handle_request_material_action(
                             })
                         }
                         Err(e) => {
-                            error!("[modify_material] failed for id={}: {}", modify_payload.id, e);
+                            error!(
+                                "[modify_material] failed for id={}: {}",
+                                modify_payload.id, e
+                            );
                             serde_json::json!({
                                 "type": "notify",
                                 "payload": {