Skip to content

refactor: update upsert API to handle multiple rows and adjust related service methods #119

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Aadil-Ahmed-27 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

refactor: update upsert API to handle multiple rows and adjust related service methods #119

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions src/environment.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
)),
Expand Down Expand Up @@ -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),
Expand Down
56 changes: 17 additions & 39 deletions src/handlers/routes.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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(&gtfs_id.as_str());
let fetch_external = !just_internal && !INTERNAL_ONLY_GTFS_IDS.contains(&gtfs_id.as_str());
if fetch_external {
let mut ext_rows = app_state
.db_vehicle_reader
Expand All @@ -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(&gtfs_id.as_str());
let fetch_internal = !just_external && !EXTERNAL_ONLY_GTFS_IDS.contains(&gtfs_id.as_str());
if fetch_internal {
let mut int_rows = app_state
.db_vehicle_reader_internal
Expand Down Expand Up @@ -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<AppState>,
path: Path<(String, String)>,
body: Json<Value>,
Expand All @@ -3644,7 +3637,7 @@ pub async fn upsert_one_row(

let result = app_state
.operator_service
.upsert_one_row(&table, &gtfs_id, body_value, to_regen)
.upsert_many_rows(&table, &gtfs_id, body_value, to_regen)
.await?;

Ok(HttpResponse::Ok().json(result))
Expand Down Expand Up @@ -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(&gtfs_id, &query.q, limit, query.with_routes.unwrap_or(false))
.search_stops(
&gtfs_id,
&query.q,
limit,
query.with_routes.unwrap_or(false),
)
.await?;
Ok(HttpResponse::Ok().json(res))
}
Expand Down Expand Up @@ -3835,7 +3833,11 @@ pub async fn reprocess_routes(
let body = body.into_inner();
let res = app_state
.operator_service
.reprocess_routes(&gtfs_id, &body.route_ids, body.recompute_polyline.unwrap_or(false))
.reprocess_routes(
&gtfs_id,
&body.route_ids,
body.recompute_polyline.unwrap_or(false),
)
.await?;
Ok(HttpResponse::Ok().json(res))
}
Expand Down Expand Up @@ -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<AppState>,
path: Path<String>,
body: Json<EmployeeRegisterRequest>,
) -> AppResult<HttpResponse> {
let gtfs_id = path.into_inner();
let req = body.into_inner();

let response = app_state
.fleet_operator_service
.register(&gtfs_id, &req)
.await?;

Ok(HttpResponse::Ok().json(response))
}
6 changes: 3 additions & 3 deletions src/models.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,9 +108,9 @@ impl<'r> sqlx::Decode<'r, sqlx::Postgres> for IdValue {
) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
use sqlx::{TypeInfo, ValueRef};
match value.type_info().name() {
"INT8" => Ok(IdValue::Int(
<i64 as sqlx::Decode<sqlx::Postgres>>::decode(value)?,
)),
"INT8" => Ok(IdValue::Int(<i64 as sqlx::Decode<sqlx::Postgres>>::decode(
value,
)?)),
"INT4" => Ok(IdValue::Int(
<i32 as sqlx::Decode<sqlx::Postgres>>::decode(value)? as i64,
)),
Expand Down
128 changes: 0 additions & 128 deletions src/services/fleet_operator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,6 @@ pub struct EmployeeLoginResponse {
pub token: Option<String>,
pub role: Option<Role>,
}

/// 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.
Expand All @@ -147,21 +146,6 @@ fn login_response_from_row(row: Option<(String, Option<String>)>) -> 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<Role>,
}

#[derive(Debug, Clone, Deserialize, Serialize, utoipa::ToSchema)]
pub struct EmployeeRegisterResponse {
pub success: bool,
pub token_no: String,
}

// ─── Service trait ─────────────────────────────────────────────────────────────

#[async_trait]
Expand Down Expand Up @@ -207,12 +191,6 @@ pub trait FleetOperatorService: Send + Sync {
gtfs_id: &str,
req: &EmployeeLoginRequest,
) -> AppResult<EmployeeLoginResponse>;

async fn register(
&self,
gtfs_id: &str,
req: &EmployeeRegisterRequest,
) -> AppResult<EmployeeRegisterResponse>;
}

// ─── Mock implementation ───────────────────────────────────────────────────────
Expand Down Expand Up @@ -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<EmployeeRegisterResponse> {
Err(AppError::NotFound(
"Database is not connected in local testing mode.".to_string(),
))
}
}

// ─── DB implementation ─────────────────────────────────────────────────────────
Expand Down Expand Up @@ -1299,102 +1267,6 @@ impl FleetOperatorService for DBFleetOperatorService {
}),
}
}

async fn register(
&self,
gtfs_id: &str,
req: &EmployeeRegisterRequest,
) -> AppResult<EmployeeRegisterResponse> {
let designation_id: Option<String> = match req.role {
Some(role) => {
let name = match role {
Role::Driver => "driver",
Role::Conductor => "conductor",
};
let id: Option<String> = 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)]
Expand Down
Loading