From e062af14d75828d2799ea033c3b4ba0362268dcc Mon Sep 17 00:00:00 2001 From: Pakin Date: Fri, 3 Jul 2026 15:52:44 +0700 Subject: [PATCH] feat: recipe repo update routine - wip: firmware service - wip: remote shell Signed-off-by: Pakin --- src/app.rs | 204 ++++++++++++++++++++++++++++++-- src/websocket/handler.rs | 15 ++- src/websocket/model.rs | 74 ++++++++++-- src/websocket/rw.rs | 41 ++++--- src/websocket/session.rs | 3 +- src/websocket/tasks/firmware.rs | 75 ++++++++++++ src/websocket/tasks/mod.rs | 1 + src/websocket/tasks/price.rs | 15 ++- src/websocket/tasks/recipe.rs | 92 ++++++++++---- 9 files changed, 451 insertions(+), 69 deletions(-) create mode 100644 src/websocket/tasks/firmware.rs diff --git a/src/app.rs b/src/app.rs index e07fbd9..f3c2a89 100644 --- a/src/app.rs +++ b/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>>, pub shared_configures: Arc>, pub allowed_origins: Vec, @@ -46,6 +50,7 @@ impl DevConfig { domain: String, rp_service: String, api_redis_url: String, + api_git_taobin_project: String, api_sheet_endpoints: Arc>>, shared_configures: Arc>, ) -> 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 { 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 = { + 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 = @@ -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_pull_sync_request( +pub async fn invoke_taobin_project_checkout_request( http_client: &reqwest::Client, config: DevConfig, + path: String, ) -> Result> { + 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> { 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> { + 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, +) -> Result<(), Box> { + 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> { 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> { 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) = - tokio_postgres::connect(&postgres_connection_config, NoTls).await?; + let (client, connection) = tokio_postgres::connect(&postgres_connection_config, NoTls).await?; tokio::spawn(async move { if let Err(e) = connection.await { error!("connection postgres error: {e}"); diff --git a/src/websocket/handler.rs b/src/websocket/handler.rs index f52e30b..aea5106 100644 --- a/src/websocket/handler.rs +++ b/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); } { diff --git a/src/websocket/model.rs b/src/websocket/model.rs index a20f59f..5c93506 100644 --- a/src/websocket/model.rs +++ b/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, + + /// 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, +} + +/// Request to start finalize for building firmware. +#[derive(Debug, Serialize, Deserialize)] +pub struct FinalizeFirmwareRequest { + pub country: String, + pub requester_info: serde_json::Value, +} diff --git a/src/websocket/rw.rs b/src/websocket/rw.rs index 018f47c..9bba5c4 100644 --- a/src/websocket/rw.rs +++ b/src/websocket/rw.rs @@ -180,21 +180,24 @@ pub async fn read( continue; } } - "log_report" if let Some(log_payload) = req.payload => { - let log_report_payload: LogReportPayload = - match safe_deserialize(&log_payload) { - Ok(lreq) => lreq, - Err(e) => { - error!( - "error deserialize body log request: {e:?} ---> Skip" - ); - continue; - } - }; - // generate timestamp - // - let now = Instant::now(); - } + // NOTE: discontinued as it is unused and not planned to use + // + // + // "log_report" if let Some(log_payload) = req.payload => { + // let log_report_payload: LogReportPayload = + // match safe_deserialize(&log_payload) { + // Ok(lreq) => lreq, + // Err(e) => { + // error!( + // "error deserialize body log request: {e:?} ---> Skip" + // ); + // continue; + // } + // }; + // // 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 diff --git a/src/websocket/session.rs b/src/websocket/session.rs index b48353a..f1caa23 100644 --- a/src/websocket/session.rs +++ b/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, // See the version of client, as 0.0.2 is start of using secured but not sending this field yet } #[derive(serde::Serialize)] diff --git a/src/websocket/tasks/firmware.rs b/src/websocket/tasks/firmware.rs new file mode 100644 index 0000000..8f2e356 --- /dev/null +++ b/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, +} + +pub async fn handle_request_firmware( + http_client: &reqwest::Client, + config: DevConfig, + mut redis: redis::Client, + tx: Sender, + req: WebsocketMessageRequest, + uid_clone: Arc>, +) -> 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(()) +} diff --git a/src/websocket/tasks/mod.rs b/src/websocket/tasks/mod.rs index 15c63fe..b54e2c0 100644 --- a/src/websocket/tasks/mod.rs +++ b/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; diff --git a/src/websocket/tasks/price.rs b/src/websocket/tasks/price.rs index e728aeb..1cc113b 100644 --- a/src/websocket/tasks/price.rs +++ b/src/websocket/tasks/price.rs @@ -148,11 +148,16 @@ pub async fn handle_price_request( let price_action = price_param.action; - let price_content = - match invoke_checkout_request(http_client, config.clone(), price_file_format.clone()).await { - Ok(pc) => pc, - Err(e) => return Err(format!("Cannot find price of expected country: {e:?}").into()), - }; + let price_content = match invoke_recipe_checkout_request( + http_client, + config.clone(), + 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()); diff --git a/src/websocket/tasks/recipe.rs b/src/websocket/tasks/recipe.rs index d4d296f..9e260a2 100644 --- a/src/websocket/tasks/recipe.rs +++ b/src/websocket/tasks/recipe.rs @@ -73,7 +73,7 @@ async fn get_latest_recipe_from_git( ) -> Result> { 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 { - Ok(file_content) => file_content, - Err(e) => { - println!("Error on checkout: {e}"); - "".to_string() - } - }; + let content = + match invoke_recipe_checkout_request(http_client, config.clone(), r1_key).await { + Ok(file_content) => file_content, + Err(e) => { + println!("Error on checkout: {e}"); + "".to_string() + } + }; info!("[get-latest] content ready: {}", content.len()); let recipe = serde_json::from_str::(&content); @@ -116,7 +117,8 @@ async fn get_latest_recipe_saved_machine_from_git( ) -> Result> { 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, ¶ms).await { Ok(rows) => Ok(rows), @@ -1208,9 +1242,14 @@ pub async fn handle_request_material_action( && let Ok(modify_payload) = serde_json::from_value::(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": {