diff --git a/src/environment.rs b/src/environment.rs index 7356b9a..55f5220 100644 --- a/src/environment.rs +++ b/src/environment.rs @@ -207,7 +207,11 @@ impl AppState { Ok(internal_pool) => { info!("Internal database pool created successfully."); ( - Arc::new(DBOperatorService::new(internal_pool.clone(), app_config.osrm_url.clone(), app_config.gen_int_for_id.unwrap_or(false))), + Arc::new(DBOperatorService::new( + internal_pool.clone(), + app_config.osrm_url.clone(), + app_config.gen_int_for_id.unwrap_or(false), + )), Arc::new(DBVehicleReaderInternal::new( internal_pool.clone(), )), @@ -287,7 +291,11 @@ impl AppState { anyhow::anyhow!("Failed to create internal database pool: {}", e) })?; ( - Arc::new(DBOperatorService::new(internal_pool.clone(), app_config.osrm_url.clone(), app_config.gen_int_for_id.unwrap_or(false))), + Arc::new(DBOperatorService::new( + internal_pool.clone(), + app_config.osrm_url.clone(), + app_config.gen_int_for_id.unwrap_or(false), + )), Arc::new(DBVehicleReaderInternal::new(internal_pool.clone())), Arc::new(DBFleetOperatorService::new(internal_pool.clone())), Some(internal_pool), diff --git a/src/handlers/routes.rs b/src/handlers/routes.rs index cc8edb8..a99d8b0 100644 --- a/src/handlers/routes.rs +++ b/src/handlers/routes.rs @@ -1,6 +1,5 @@ use crate::services::fleet_operator::{ - EmployeeLoginRequest, EmployeeLoginResponse, EmployeeRegisterRequest, EmployeeRegisterResponse, - TripAction, WaybillAnchor, + EmployeeLoginRequest, EmployeeLoginResponse, TripAction, WaybillAnchor, }; use crate::services::operator::{ break_types, day_types, shift_types, trip_types, waybill_statuses, QueryBody, @@ -107,7 +106,7 @@ pub fn create_routes(cfg: &mut actix_web::web::ServiceConfig) { .route("/crud/{table}/all", web::get().to(get_all_rows)) .route("/crud/{table}/query", web::post().to(query_rows_handler)) .route("/crud/{table}/delete", web::post().to(delete_one_row)) - .route("/crud/{table}/upsert", web::post().to(upsert_one_row)) + .route("/crud/{table}/upsert", web::post().to(upsert_many_rows)) .route("/service-types", web::get().to(get_service_types)) .route("/routes", web::get().to(get_operator_routes)) .route("/depots", web::get().to(get_depots)) @@ -158,10 +157,6 @@ pub fn create_routes(cfg: &mut actix_web::web::ServiceConfig) { .route( "/employee/login", web::post().to(fleet_operator_employee_login), - ) - .route( - "/employee/register", - web::post().to(fleet_operator_employee_register), ), ) .route( @@ -2998,8 +2993,7 @@ pub async fn get_bus_route_schedule( // Fetch from external tables unless the query forces internal-only // or the gtfs_id is listed as internal-only. - let fetch_external = - !just_internal && !INTERNAL_ONLY_GTFS_IDS.contains(>fs_id.as_str()); + let fetch_external = !just_internal && !INTERNAL_ONLY_GTFS_IDS.contains(>fs_id.as_str()); if fetch_external { let mut ext_rows = app_state .db_vehicle_reader @@ -3010,8 +3004,7 @@ pub async fn get_bus_route_schedule( // Fetch from internal tables unless the query forces external-only // or the gtfs_id is listed as external-only. - let fetch_internal = - !just_external && !EXTERNAL_ONLY_GTFS_IDS.contains(>fs_id.as_str()); + let fetch_internal = !just_external && !EXTERNAL_ONLY_GTFS_IDS.contains(>fs_id.as_str()); if fetch_internal { let mut int_rows = app_state .db_vehicle_reader_internal @@ -3618,7 +3611,7 @@ pub async fn delete_one_row( request_body = serde_json::Value, responses((status = 200, description = "Row upserted")) )] -pub async fn upsert_one_row( +pub async fn upsert_many_rows( app_state: Data, path: Path<(String, String)>, body: Json, @@ -3644,7 +3637,7 @@ pub async fn upsert_one_row( let result = app_state .operator_service - .upsert_one_row(&table, >fs_id, body_value, to_regen) + .upsert_many_rows(&table, >fs_id, body_value, to_regen) .await?; Ok(HttpResponse::Ok().json(result)) @@ -3708,7 +3701,12 @@ pub async fn search_stops( let limit = query.limit.unwrap_or(20).clamp(1, 200); let res = app_state .operator_service - .search_stops(>fs_id, &query.q, limit, query.with_routes.unwrap_or(false)) + .search_stops( + >fs_id, + &query.q, + limit, + query.with_routes.unwrap_or(false), + ) .await?; Ok(HttpResponse::Ok().json(res)) } @@ -3835,7 +3833,11 @@ pub async fn reprocess_routes( let body = body.into_inner(); let res = app_state .operator_service - .reprocess_routes(>fs_id, &body.route_ids, body.recompute_polyline.unwrap_or(false)) + .reprocess_routes( + >fs_id, + &body.route_ids, + body.recompute_polyline.unwrap_or(false), + ) .await?; Ok(HttpResponse::Ok().json(res)) } @@ -4513,27 +4515,3 @@ pub async fn fleet_operator_employee_login( Ok(HttpResponse::Ok().json(response)) } - -#[utoipa::path( - post, - path = "/internal/fleet-operator/{gtfs_id}/employee/register", - tag = "Fleet Operator", - params(("gtfs_id" = String, Path, description = "GTFS dataset identifier")), - request_body = EmployeeRegisterRequest, - responses((status = 200, description = "Registration response", body = EmployeeRegisterResponse)) -)] -pub async fn fleet_operator_employee_register( - app_state: Data, - path: Path, - body: Json, -) -> AppResult { - let gtfs_id = path.into_inner(); - let req = body.into_inner(); - - let response = app_state - .fleet_operator_service - .register(>fs_id, &req) - .await?; - - Ok(HttpResponse::Ok().json(response)) -} diff --git a/src/models.rs b/src/models.rs index 690566f..614b0cf 100644 --- a/src/models.rs +++ b/src/models.rs @@ -108,9 +108,9 @@ impl<'r> sqlx::Decode<'r, sqlx::Postgres> for IdValue { ) -> Result> { use sqlx::{TypeInfo, ValueRef}; match value.type_info().name() { - "INT8" => Ok(IdValue::Int( - >::decode(value)?, - )), + "INT8" => Ok(IdValue::Int(>::decode( + value, + )?)), "INT4" => Ok(IdValue::Int( >::decode(value)? as i64, )), diff --git a/src/services/fleet_operator.rs b/src/services/fleet_operator.rs index 9c9bf78..404f506 100644 --- a/src/services/fleet_operator.rs +++ b/src/services/fleet_operator.rs @@ -121,7 +121,6 @@ pub struct EmployeeLoginResponse { pub token: Option, pub role: Option, } - /// Map a matched `(token_no, designation_name)` login row into a response. Shared by /// every `auth_type`: once an employee row is found — by whichever credential — deriving /// the badge token and role is identical, so only the lookup query differs per auth type. @@ -147,21 +146,6 @@ fn login_response_from_row(row: Option<(String, Option)>) -> EmployeeLog } } -#[derive(Debug, Clone, Deserialize, Serialize, utoipa::ToSchema)] -pub struct EmployeeRegisterRequest { - pub token_no: String, - pub email_hash: String, - pub password_hash: String, - pub first_name: String, - pub role: Option, -} - -#[derive(Debug, Clone, Deserialize, Serialize, utoipa::ToSchema)] -pub struct EmployeeRegisterResponse { - pub success: bool, - pub token_no: String, -} - // ─── Service trait ───────────────────────────────────────────────────────────── #[async_trait] @@ -207,12 +191,6 @@ pub trait FleetOperatorService: Send + Sync { gtfs_id: &str, req: &EmployeeLoginRequest, ) -> AppResult; - - async fn register( - &self, - gtfs_id: &str, - req: &EmployeeRegisterRequest, - ) -> AppResult; } // ─── Mock implementation ─────────────────────────────────────────────────────── @@ -298,16 +276,6 @@ impl FleetOperatorService for MockFleetOperatorService { "Database is not connected in local testing mode.".to_string(), )) } - - async fn register( - &self, - _gtfs_id: &str, - _req: &EmployeeRegisterRequest, - ) -> AppResult { - Err(AppError::NotFound( - "Database is not connected in local testing mode.".to_string(), - )) - } } // ─── DB implementation ───────────────────────────────────────────────────────── @@ -1299,102 +1267,6 @@ impl FleetOperatorService for DBFleetOperatorService { }), } } - - async fn register( - &self, - gtfs_id: &str, - req: &EmployeeRegisterRequest, - ) -> AppResult { - let designation_id: Option = match req.role { - Some(role) => { - let name = match role { - Role::Driver => "driver", - Role::Conductor => "conductor", - }; - let id: Option = sqlx::query_scalar( - r#" - SELECT designation_id - FROM designations_internal - WHERE LOWER(designation_name) = $1 - AND gtfs_id = $2 - AND deleted = false - LIMIT 1 - "#, - ) - .bind(name) - .bind(gtfs_id) - .fetch_optional(&self.pool) - .await - .map_err(|e| { - error!( - "register designation lookup failed for gtfs_id={}, role={}: {}", - gtfs_id, name, e - ); - AppError::Internal(e.to_string()) - })?; - Some(id.ok_or_else(|| { - AppError::NotFound(format!( - "designation '{}' not found for gtfs_id={}", - name, gtfs_id - )) - })?) - } - None => None, - }; - - if let Some(did) = designation_id { - sqlx::query( - r#" - INSERT INTO employees_internal (token_no, email_hash, password_hash, gtfs_id, first_name, designation_id) - VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT (gtfs_id, token_no) DO UPDATE SET - email_hash = EXCLUDED.email_hash, - password_hash = EXCLUDED.password_hash, - designation_id = EXCLUDED.designation_id, - updated_at = NOW() - "#, - ) - .bind(&req.token_no) - .bind(&req.email_hash) - .bind(&req.password_hash) - .bind(gtfs_id) - .bind(&req.first_name) - .bind(did) - .execute(&self.pool) - .await - .map_err(|e| { - error!("register upsert failed for gtfs_id={}, token_no={}: {}", gtfs_id, req.token_no, e); - AppError::Internal(e.to_string()) - })?; - } else { - sqlx::query( - r#" - INSERT INTO employees_internal (token_no, email_hash, password_hash, gtfs_id, first_name) - VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (gtfs_id, token_no) DO UPDATE SET - email_hash = EXCLUDED.email_hash, - password_hash = EXCLUDED.password_hash, - updated_at = NOW() - "#, - ) - .bind(&req.token_no) - .bind(&req.email_hash) - .bind(&req.password_hash) - .bind(gtfs_id) - .bind(&req.first_name) - .execute(&self.pool) - .await - .map_err(|e| { - error!("register upsert failed for gtfs_id={}, token_no={}: {}", gtfs_id, req.token_no, e); - AppError::Internal(e.to_string()) - })?; - } - - Ok(EmployeeRegisterResponse { - success: true, - token_no: req.token_no.clone(), - }) - } } #[cfg(test)] diff --git a/src/services/operator.rs b/src/services/operator.rs index 37b633f..743c4c5 100644 --- a/src/services/operator.rs +++ b/src/services/operator.rs @@ -256,11 +256,13 @@ pub fn table_columns(table: &str) -> Option<&'static [&'static str]> { "driving_license_expiry", "driving_license_number", "email", + "email_hash", "father_name", "first_name", "gender", "last_name", "mobile_no", + "password_hash", "status", "token_no", "updated_at", @@ -396,7 +398,7 @@ pub fn table_pk(table: &str) -> Option<&'static str> { "service_type_internal" => Some("service_type_id"), "stop_internal" => Some("bus_stop_id"), "designations_internal" => Some("designation_id"), - "employees_internal" => Some("emp_id"), + "employees_internal" => Some("gtfs_id, token_no"), "entities_internal" => Some("entity_id"), "vehicles_internal" => Some("vehicle_id"), "waybill_device_internal" => Some("waybill_device_id"), @@ -682,7 +684,10 @@ async fn enrich_stops( /// Call OSRM `/route` with all stop coords (lat, lon) → (encoded polyline, per-leg (distance_m, duration_s)). /// Returns None if `base` is None/empty, fewer than 2 coords, or any request/parse error. -async fn osrm_route(base: Option<&str>, coords: &[(f64, f64)]) -> Option<(String, Vec<(f64, f64)>)> { +async fn osrm_route( + base: Option<&str>, + coords: &[(f64, f64)], +) -> Option<(String, Vec<(f64, f64)>)> { let base = base.filter(|s| !s.is_empty())?; if coords.len() < 2 { return None; @@ -697,13 +702,12 @@ async fn osrm_route(base: Option<&str>, coords: &[(f64, f64)]) -> Option<(String base.trim_end_matches('/'), coord_str ); - let json: serde_json::Value = - tokio::time::timeout(Duration::from_secs(5), async { - reqwest::get(&url).await?.json::().await - }) - .await - .ok()? - .ok()?; + let json: serde_json::Value = tokio::time::timeout(Duration::from_secs(5), async { + reqwest::get(&url).await?.json::().await + }) + .await + .ok()? + .ok()?; let route0 = json.get("routes")?.as_array()?.first()?; let geometry = route0.get("geometry")?.as_str()?.to_string(); let legs = route0 @@ -1181,7 +1185,7 @@ pub trait OperatorService: Send + Sync { async fn delete_one_row(&self, table: &str, gtfs_id: &str, data: Value) -> AppResult; - async fn upsert_one_row( + async fn upsert_many_rows( &self, table: &str, gtfs_id: &str, @@ -1280,11 +1284,8 @@ pub trait OperatorService: Send + Sync { ) -> AppResult; /// Route points joined with stop details, ordered by route_order (editor feed). - async fn get_route_stops( - &self, - gtfs_id: &str, - route_id: &str, - ) -> AppResult; + async fn get_route_stops(&self, gtfs_id: &str, route_id: &str) + -> AppResult; /// Insert a stop at position `position`, shifting existing route_order up by 1. async fn insert_route_stop( @@ -1304,10 +1305,7 @@ pub trait OperatorService: Send + Sync { ) -> AppResult>; /// Full route-stop-mapping across all routes (for CSV export). - async fn export_route_stop_mapping( - &self, - gtfs_id: &str, - ) -> AppResult>; + async fn export_route_stop_mapping(&self, gtfs_id: &str) -> AppResult>; /// Returns the underlying pool for streaming responses. None for mock implementations. fn pool(&self) -> Option<&PgPool> { @@ -1362,7 +1360,7 @@ impl OperatorService for MockOperatorService { mock_err!() } - async fn upsert_one_row( + async fn upsert_many_rows( &self, _table: &str, _gtfs_id: &str, @@ -1531,10 +1529,7 @@ impl OperatorService for MockOperatorService { mock_err!() } - async fn export_route_stop_mapping( - &self, - _gtfs_id: &str, - ) -> AppResult> { + async fn export_route_stop_mapping(&self, _gtfs_id: &str) -> AppResult> { mock_err!() } } @@ -1881,30 +1876,72 @@ impl OperatorService for DBOperatorService { .as_object() .ok_or_else(|| AppError::BadRequest("Body must be a JSON object".to_string()))?; - let pk_value = obj.get(pk).ok_or_else(|| { - AppError::BadRequest(format!("Body must contain the primary key: {}", pk)) - })?; - - let id = match pk_value { - Value::Number(n) => n - .as_i64() - .map(|i| i.to_string()) - .or_else(|| n.as_f64().map(|f| f.to_string())) - .ok_or_else(|| AppError::BadRequest("ID must be a number".to_string()))?, - Value::String(s) => s.clone(), - _ => return Err(AppError::BadRequest("ID must be a string or number".to_string())), + let key_cols: Vec<&str> = pk.split(',').map(|s| s.trim()).collect(); + + let mut where_clauses = Vec::new(); + let mut bind_index = 1; + + for key_col in &key_cols { + if *key_col == "gtfs_id" { + where_clauses.push(format!("gtfs_id = ${}", bind_index)); + } else { + obj.get(*key_col).ok_or_else(|| { + AppError::BadRequest(format!("Body must contain key column: {}", key_col)) + })?; + // `::text` cast so the id predicate works whether the column is still + // `bigint` (prod) or already migrated to `text` — bind the string either way. + where_clauses.push(format!("{}::text = ${}", key_col, bind_index)); + } + bind_index += 1; + } + + let sql = if key_cols.contains(&"gtfs_id") { + format!( + "UPDATE public.{} SET deleted = true, updated_at = now() WHERE {} AND deleted = false", + table, + where_clauses.join(" AND ") + ) + } else { + format!( + "UPDATE public.{} SET deleted = true, updated_at = now() WHERE {} AND gtfs_id = ${} AND deleted = false", + table, + where_clauses.join(" AND "), + bind_index + ) }; - // `::text` cast so the id predicate works whether the column is still - // `bigint` (prod) or already migrated to `text` — bind the string either way. - let sql = format!( - "UPDATE public.{} SET deleted = true, updated_at = now() WHERE {}::text = $1 AND gtfs_id = $2 AND deleted = false", - table, pk - ); + let mut q = sqlx::query(&sql); - let result = sqlx::query(&sql) - .bind(&id) - .bind(gtfs_id) + for key_col in &key_cols { + if *key_col == "gtfs_id" { + q = q.bind(gtfs_id); + } else { + let val = obj.get(*key_col).unwrap(); + let s = match val { + Value::Number(n) => n + .as_i64() + .map(|i| i.to_string()) + .or_else(|| n.as_f64().map(|f| f.to_string())) + .ok_or_else(|| { + AppError::BadRequest(format!("Invalid number for {}", key_col)) + })?, + Value::String(s) => s.clone(), + _ => { + return Err(AppError::BadRequest(format!( + "Invalid type for {}", + key_col + ))) + } + }; + q = q.bind(s); + } + } + + if !key_cols.contains(&"gtfs_id") { + q = q.bind(gtfs_id); + } + + let result = q .execute(&self.pool) .await .map_err(|e| AppError::DbError(format!("delete_one_row {}: {}", table, e)))?; @@ -1912,7 +1949,7 @@ impl OperatorService for DBOperatorService { Ok(result.rows_affected()) } - async fn upsert_one_row( + async fn upsert_many_rows( &self, table: &str, gtfs_id: &str, @@ -1938,18 +1975,26 @@ impl OperatorService for DBOperatorService { return Err(AppError::BadRequest("Body is empty".to_string())); } - // Auto-inject PK with a random ID if missing or null + // Auto-inject PK with a random ID if missing or null. `pk` may be a comma-separated + // composite (e.g. "gtfs_id, token_no"); inject per non-gtfs_id key column so the + // key column names actually exist on the row (gtfs_id is bound from the URL). + let pk_key_cols: Vec<&str> = pk.split(',').map(|s| s.trim()).collect(); for val in arr.iter_mut() { if let Some(obj) = val.as_object_mut() { - let missing = obj.get(pk).map_or(true, |v| v.is_null()); - if missing { - // numeric id for still-`bigint` columns (pre-migration), else UUID - let id = if self.gen_int_for_id { - Value::from(field_generator::gen_random_int_id()) - } else { - Value::String(field_generator::gen_random_id()) - }; - obj.insert(pk.to_string(), id); + for key_col in &pk_key_cols { + if *key_col == "gtfs_id" { + continue; + } + let missing = obj.get(*key_col).map_or(true, |v| v.is_null()); + if missing { + // numeric id for still-`bigint` columns (pre-migration), else UUID + let id = if self.gen_int_for_id { + Value::from(field_generator::gen_random_int_id()) + } else { + Value::String(field_generator::gen_random_id()) + }; + obj.insert(key_col.to_string(), id); + } } } } @@ -1960,7 +2005,9 @@ impl OperatorService for DBOperatorService { // read path requires a non-null schedule_number, so we backfill here for all callers. if table == "bus_schedule_trip_detail_internal" { for val in arr.iter_mut() { - let Some(obj) = val.as_object_mut() else { continue }; + let Some(obj) = val.as_object_mut() else { + continue; + }; let needs_enrich = ["schedule_number", "org_name", "shift_type_name"] .iter() .any(|k| obj.get(*k).map_or(true, |v| v.is_null())); @@ -2007,33 +2054,59 @@ impl OperatorService for DBOperatorService { return Err(AppError::BadRequest("First object is empty".to_string())); } - // Build column list from caller's keys, excluding gtfs_id (we inject it ourselves) - // Also exclude keys whose value is JSON null — omit-null semantics so that a partial - // update (e.g. just {waybill_id, duty_date}) never triggers NOT NULL violations on - // columns the caller didn't intend to touch. - let mut cols: Vec<&str> = first_obj - .keys() - .filter(|k| k.as_str() != "gtfs_id" && !first_obj[k.as_str()].is_null()) - .map(|s| s.as_str()) - .collect(); + // Build column list as the UNION of non-null keys across all rows, excluding gtfs_id + // (we inject it ourselves). For each row × col, the VALUES clause either binds the + // row's value or emits the literal `DEFAULT` keyword for missing/null cells; Postgres + // resolves `DEFAULT` to the column's declared default (or NULL if none). This lets a + // single bulk INSERT handle heterogeneous row shapes without binding NULL for missing + // cells (which would violate NOT NULL on fresh INSERTs). + // + // On ON CONFLICT, `COALESCE(EXCLUDED.col, table.col)` preserves the existing value + // when EXCLUDED.col resolved to NULL (the typical "this row didn't touch this col" + // case for nullable columns without DB defaults). + let mut col_set: std::collections::HashSet<&str> = std::collections::HashSet::new(); + for row in &arr { + if let Some(obj) = row.as_object() { + for (k, v) in obj.iter() { + if k.as_str() != "gtfs_id" && !v.is_null() { + col_set.insert(k.as_str()); + } + } + } + } + let mut cols: Vec<&str> = col_set.into_iter().collect(); + cols.sort(); // Sort for consistent SQL generation cols.push("gtfs_id"); // gtfs_id always last + // Parse primary key columns and exclude them from UPDATE SET + let key_cols: Vec<&str> = pk.split(',').map(|s| s.trim()).collect(); + let update_set: Vec = cols .iter() - .map(|c| format!("{} = EXCLUDED.{}", c, c)) + .filter(|c| !key_cols.contains(c)) + .map(|c| format!("{} = COALESCE(EXCLUDED.{}, {}.{})", c, c, table, c)) .collect(); - let mut placeholders = Vec::new(); - let mut bind_index = 1; + // Build per-row placeholder tuples in lockstep with the binding plan: each cell is + // either "$N" (value will be bound below) or the literal "DEFAULT". + let mut placeholders: Vec = Vec::with_capacity(arr.len()); + let mut bind_index: usize = 1; - for _ in 0..arr.len() { - let row_placeholders: Vec = (0..cols.len()) - .map(|_| { - let s = format!("${}", bind_index); + for val in arr.iter() { + let obj = val.as_object().ok_or_else(|| { + AppError::BadRequest("Array must contain JSON objects".to_string()) + })?; + let mut row_placeholders: Vec = Vec::with_capacity(cols.len()); + for col in cols.iter() { + let bind_cell = + *col == "gtfs_id" || matches!(obj.get(*col), Some(v) if !v.is_null()); + if bind_cell { + row_placeholders.push(format!("${}", bind_index)); bind_index += 1; - s - }) - .collect(); + } else { + row_placeholders.push("DEFAULT".to_string()); + } + } placeholders.push(format!("({})", row_placeholders.join(", "))); } @@ -2059,7 +2132,7 @@ impl OperatorService for DBOperatorService { .bind(table) .fetch_all(&self.pool) .await - .map_err(|e| AppError::DbError(format!("upsert_one_row col types {}: {}", table, e)))? + .map_err(|e| AppError::DbError(format!("upsert_many_rows col types {}: {}", table, e)))? .into_iter() .collect(); let is_int_col = |c: &str| { @@ -2071,13 +2144,20 @@ impl OperatorService for DBOperatorService { let mut q = sqlx::query_scalar::<_, Value>(&sql); - for val in &arr { - let obj = val.as_object().ok_or_else(|| { - AppError::BadRequest("Array must contain JSON objects".to_string()) - })?; - // bind all cols except gtfs_id first - for col in cols.iter().filter(|c| **c != "gtfs_id") { - let col_val = obj.get(*col).unwrap_or(&Value::Null); + // Bind in the exact same order the placeholders were emitted. Cells where we wrote + // "DEFAULT" above contribute no bind here. + for val in arr.iter() { + // Safe: arr was validated to contain only JSON objects in the placeholder pass. + let obj = val.as_object().unwrap(); + for col in cols.iter() { + if *col == "gtfs_id" { + q = q.bind(gtfs_id); + continue; + } + let col_val = match obj.get(*col) { + Some(v) if !v.is_null() => v, + _ => continue, // emitted DEFAULT above + }; let int_col = is_int_col(col); match col_val { Value::String(s) => { @@ -2139,22 +2219,20 @@ impl OperatorService for DBOperatorService { } } Value::Bool(b) => q = q.bind(b), - Value::Null => q = q.bind(Option::::None), _ => { - return Err(AppError::BadRequest( - "Unsupported JSON value type for key".to_string(), - )) + return Err(AppError::BadRequest(format!( + "Unsupported JSON value type for column {}", + col + ))) } } } - // inject gtfs_id last - q = q.bind(gtfs_id); } let results = q .fetch_all(&self.pool) .await - .map_err(|e| AppError::DbError(format!("upsert_one_row {}: {}", table, e)))?; + .map_err(|e| AppError::DbError(format!("upsert_many_rows {}: {}", table, e)))?; let ret = if is_array { Value::Array(results) @@ -2743,7 +2821,10 @@ impl OperatorService for DBOperatorService { let stage_name = data.get("stage_name").and_then(|v| v.as_str()); let travel_time = data.get("travel_time").and_then(|v| v.as_str()); // New route stops are visible and active by default. - let is_visible = data.get("is_visible").and_then(|v| v.as_bool()).unwrap_or(true); + let is_visible = data + .get("is_visible") + .and_then(|v| v.as_bool()) + .unwrap_or(true); let new_id = field_generator::gen_random_id(); let mut tx = self @@ -2936,7 +3017,10 @@ impl OperatorService for DBOperatorService { polyline_out = Some(geom); } None => { - warn!("reprocess_routes: OSRM unavailable for route {}, polyline not updated", route_id); + warn!( + "reprocess_routes: OSRM unavailable for route {}, polyline not updated", + route_id + ); } } } diff --git a/src/swagger.rs b/src/swagger.rs index eb80c50..81538c4 100644 --- a/src/swagger.rs +++ b/src/swagger.rs @@ -2,9 +2,7 @@ use utoipa::OpenApi; use crate::handlers::routes; use crate::models; -use crate::services::fleet_operator::{ - EmployeeLoginRequest, EmployeeLoginResponse, EmployeeRegisterRequest, EmployeeRegisterResponse, -}; +use crate::services::fleet_operator::{EmployeeLoginRequest, EmployeeLoginResponse}; use crate::services::operator::QueryBody; #[derive(OpenApi)] @@ -76,7 +74,7 @@ use crate::services::operator::QueryBody; routes::get_all_rows, routes::query_rows_handler, routes::delete_one_row, - routes::upsert_one_row, + routes::upsert_many_rows, routes::get_service_types, routes::get_operator_routes, routes::get_depots, @@ -103,7 +101,6 @@ use crate::services::operator::QueryBody; routes::fleet_operator_current_trip_details, routes::fleet_operator_verify, routes::fleet_operator_employee_login, - routes::fleet_operator_employee_register, routes::get_waybill_metadata, ), components(schemas( @@ -148,8 +145,6 @@ use crate::services::operator::QueryBody; routes::FleetVerifyRequest, EmployeeLoginRequest, EmployeeLoginResponse, - EmployeeRegisterRequest, - EmployeeRegisterResponse, crate::services::fleet_operator::AuthType, crate::services::fleet_operator::Role, QueryBody,