Skip to content

feat: MobileNumber auth + optional metadata on /employee/login #130

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

feat: MobileNumber auth + optional metadata on /employee/login #130

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
59 changes: 34 additions & 25 deletions src/environment.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,18 +178,11 @@ impl AppState {
match create_database_pool(&app_config).await {
Ok(pool) => {
info!("Successfully connected to the local database.");
let vehicle_reader =
Arc::new(DBVehicleReader::new(pool.clone(), &app_config));
// operator and internal reader use ONLY the internal pool
let (
operator_svc,
vehicle_reader_internal,
fleet_op_svc,
internal_pool_opt,
): (
// Resolve internal pool first so vehicle_reader, employee_reader,
// and fleet_op_svc can all share it.
let (operator_svc, vehicle_reader_internal, internal_pool_opt): (
Arc<dyn OperatorService>,
Arc<dyn VehicleDataReaderInternal>,
Arc<dyn FleetOperatorService>,
Option<PgPool>,
) = if let Some(internal_url) = &app_config.internal_database_url {
info!("Connecting to internal database (local)...");
Expand All @@ -201,9 +194,6 @@ impl AppState {
Arc::new(DBVehicleReaderInternal::new(
internal_pool.clone(),
)),
Arc::new(DBFleetOperatorService::new(
internal_pool.clone(),
)),
Some(internal_pool),
)
}
Expand All @@ -213,8 +203,6 @@ impl AppState {
Arc::new(MockOperatorService::new()),
Arc::new(MockDBVehicleReaderInternal::new())
as Arc<dyn VehicleDataReaderInternal>,
Arc::new(MockFleetOperatorService::new())
as Arc<dyn FleetOperatorService>,
None,
)
}
Expand All @@ -225,16 +213,27 @@ impl AppState {
Arc::new(MockOperatorService::new()),
Arc::new(MockDBVehicleReaderInternal::new())
as Arc<dyn VehicleDataReaderInternal>,
Arc::new(MockFleetOperatorService::new())
as Arc<dyn FleetOperatorService>,
None,
)
};
let vehicle_reader = Arc::new(DBVehicleReader::new(
pool.clone(),
internal_pool_opt.clone(),
&app_config,
));
let employee_reader = Arc::new(DBEmployeeReader::new(
pool.clone(),
internal_pool_opt,
internal_pool_opt.clone(),
&app_config,
));
let fleet_op_svc: Arc<dyn FleetOperatorService> = match &internal_pool_opt {
Some(internal_pool) => Arc::new(DBFleetOperatorService::new(
internal_pool.clone(),
employee_reader.clone(),
vehicle_reader.clone(),
)),
None => Arc::new(MockFleetOperatorService::new()),
};
(
vehicle_reader,
employee_reader,
Expand Down Expand Up @@ -262,12 +261,11 @@ impl AppState {
let pool = create_database_pool(&app_config)
.await
.map_err(|e| anyhow::anyhow!("Failed to create database pool: {}", e))?;
let vehicle_reader = Arc::new(DBVehicleReader::new(pool.clone(), &app_config));
// operator and internal reader use ONLY the internal pool
let (operator_svc, vehicle_reader_internal, fleet_op_svc, internal_pool_opt): (
// Resolve internal pool first so vehicle_reader, employee_reader,
// and fleet_op_svc can all share it.
let (operator_svc, vehicle_reader_internal, internal_pool_opt): (
Arc<dyn OperatorService>,
Arc<dyn VehicleDataReaderInternal>,
Arc<dyn FleetOperatorService>,
Option<PgPool>,
) = if let Some(internal_url) = &app_config.internal_database_url {
info!("Connecting to internal database (production)...");
Expand All @@ -279,7 +277,6 @@ impl AppState {
(
Arc::new(DBOperatorService::new(internal_pool.clone())),
Arc::new(DBVehicleReaderInternal::new(internal_pool.clone())),
Arc::new(DBFleetOperatorService::new(internal_pool.clone())),
Some(internal_pool),
)
} else {
Expand All @@ -288,15 +285,27 @@ impl AppState {
Arc::new(MockOperatorService::new()),
Arc::new(MockDBVehicleReaderInternal::new())
as Arc<dyn VehicleDataReaderInternal>,
Arc::new(MockFleetOperatorService::new()) as Arc<dyn FleetOperatorService>,
None,
)
};
let vehicle_reader = Arc::new(DBVehicleReader::new(
pool.clone(),
internal_pool_opt.clone(),
&app_config,
));
let employee_reader = Arc::new(DBEmployeeReader::new(
pool.clone(),
internal_pool_opt,
internal_pool_opt.clone(),
&app_config,
));
let fleet_op_svc: Arc<dyn FleetOperatorService> = match &internal_pool_opt {
Some(internal_pool) => Arc::new(DBFleetOperatorService::new(
internal_pool.clone(),
employee_reader.clone(),
vehicle_reader.clone(),
)),
None => Arc::new(MockFleetOperatorService::new()),
};
(
vehicle_reader,
employee_reader,
Expand Down
15 changes: 13 additions & 2 deletions src/handlers/routes.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4235,25 +4235,36 @@ pub async fn fleet_operator_verify(
Ok(HttpResponse::Ok().json(response))
}

#[derive(Debug, Deserialize)]
pub struct EmployeeLoginQuery {
#[serde(rename = "withMetadata")]
pub with_metadata: Option<bool>,
}

#[utoipa::path(
post,
path = "/internal/fleet-operator/{gtfs_id}/employee/login",
tag = "Fleet Operator",
params(("gtfs_id" = String, Path, description = "GTFS dataset identifier")),
params(
("gtfs_id" = String, Path, description = "GTFS dataset identifier"),
("withMetadata" = Option<bool>, Query, description = "Include employee metadata (name, depot) in the response"),
),
request_body = EmployeeLoginRequest,
responses((status = 200, description = "Login response", body = EmployeeLoginResponse))
)]
pub async fn fleet_operator_employee_login(
app_state: Data<AppState>,
path: Path<String>,
query: Query<EmployeeLoginQuery>,
body: Json<EmployeeLoginRequest>,
) -> AppResult<HttpResponse> {
let gtfs_id = path.into_inner();
let req = body.into_inner();
let with_metadata = query.with_metadata.unwrap_or(false);

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

Ok(HttpResponse::Ok().json(response))
Expand Down
13 changes: 13 additions & 0 deletions src/models.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -739,6 +739,19 @@ pub struct MinimalEmployee {
pub depot_name: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct EmployeeMetadata {
pub first_name: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub last_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub mobile_no: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub depot_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub depot_code: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DepotManagerDetails {
#[serde(rename = "depotCode")]
Expand Down
170 changes: 170 additions & 0 deletions src/services/db_employee_reader.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,9 +9,34 @@ use std::time::{Duration, SystemTime};
use tokio::sync::RwLock;
use tracing::debug;

/// Minimal employee row used by mobile-number login. Depot info is resolved
/// separately via the depot cache using `entity_id`.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct EmployeeLookupRow {
pub token_no: Option<String>,
pub first_name: String,
pub last_name: Option<String>,
pub mobile_no: Option<String>,
pub entity_id: i64,
pub designation_name: Option<String>,
}

#[async_trait]
pub trait EmployeeReader: Send + Sync {
async fn get_employee_by_phone(&self, phone: &str) -> AppResult<Option<MinimalEmployee>>;

/// Look up an employee in primary `employees` joined with `designations`.
async fn lookup_by_mobile_primary(
&self,
mobile_no: &str,
) -> AppResult<Option<EmployeeLookupRow>>;

/// Look up an employee in `employees_internal` for the given gtfs_id.
async fn lookup_by_mobile_internal(
&self,
gtfs_id: &str,
mobile_no: &str,
) -> AppResult<Option<EmployeeLookupRow>>;
}

pub struct MockEmployeeReader;
Expand All @@ -35,12 +60,30 @@ impl EmployeeReader for MockEmployeeReader {
"Employee lookup disabled in mock mode".into(),
))
}

async fn lookup_by_mobile_primary(
&self,
_mobile_no: &str,
) -> AppResult<Option<EmployeeLookupRow>> {
// Mock mode has no data; report a miss rather than an error.
// Matches the MockDBVehicleReader::get_depot_info convention.
Ok(None)
}

async fn lookup_by_mobile_internal(
&self,
_gtfs_id: &str,
_mobile_no: &str,
) -> AppResult<Option<EmployeeLookupRow>> {
Ok(None)
}
}

pub struct DBEmployeeReader {
pool: PgPool,
internal_pool: Option<PgPool>,
cache: Arc<RwLock<HashMap<String, (MinimalEmployee, SystemTime)>>>,
lookup_cache: Arc<RwLock<HashMap<String, (EmployeeLookupRow, SystemTime)>>>,
cache_duration: Duration,
}

Expand All @@ -50,6 +93,7 @@ impl DBEmployeeReader {
pool,
internal_pool,
cache: Arc::new(RwLock::new(HashMap::new())),
lookup_cache: Arc::new(RwLock::new(HashMap::new())),
cache_duration: Duration::from_secs(config.cache_duration),
}
}
Expand All @@ -70,6 +114,23 @@ impl DBEmployeeReader {
let mut cache = self.cache.write().await;
cache.insert(phone.to_string(), (employee.clone(), SystemTime::now()));
}

async fn get_cached_lookup(&self, key: &str) -> Option<EmployeeLookupRow> {
let cache = self.lookup_cache.read().await;
if let Some((row, timestamp)) = cache.get(key) {
if timestamp.elapsed().unwrap_or_default() < self.cache_duration {
debug!("Cache HIT for employee lookup {}", key);
return Some(row.clone());
}
}
debug!("Cache MISS for employee lookup {}", key);
None
}

async fn cache_lookup(&self, key: &str, row: &EmployeeLookupRow) {
let mut cache = self.lookup_cache.write().await;
cache.insert(key.to_string(), (row.clone(), SystemTime::now()));
}
}

#[async_trait]
Expand Down Expand Up @@ -148,4 +209,113 @@ impl EmployeeReader for DBEmployeeReader {
))),
}
}

async fn lookup_by_mobile_primary(
&self,
mobile_no: &str,
) -> AppResult<Option<EmployeeLookupRow>> {
let cache_key = format!("primary:{}", mobile_no);
if let Some(cached) = self.get_cached_lookup(&cache_key).await {
return Ok(Some(cached));
}

// Not single-flighted: concurrent first hits for the same phone will all
// query before any writes back to the cache. Acceptable — the dupes only
// cost an extra DB round trip on a cold key.
//
// ORDER BY emp_id DESC makes the lookup deterministic when stale duplicate
// rows share a mobile_no (the schema does not enforce uniqueness): prefer
// the most recently inserted row, which is the one operations meant to
// activate.
let query = r#"
SELECT
e.token_no,
e.first_name,
e.last_name,
e.mobile_no,
e.entity_id,
LOWER(d.designation_name) AS designation_name
FROM employees e
LEFT JOIN designations d ON d.designation_id = e.designation_id
AND d.deleted = false
WHERE e.mobile_no = $1
AND e.deleted = false
ORDER BY e.emp_id DESC
LIMIT 1
"#;

match sqlx::query_as::<_, EmployeeLookupRow>(query)
.bind(mobile_no)
.fetch_optional(&self.pool)
.await
{
Ok(Some(row)) => {
debug!("Employee lookup primary HIT for mobile_no {}", mobile_no);
self.cache_lookup(&cache_key, &row).await;
Ok(Some(row))
}
Ok(None) => Ok(None),
Err(e) => Err(AppError::Internal(format!(
"DB query error (primary mobile lookup): {}",
e
))),
}
}

async fn lookup_by_mobile_internal(
&self,
gtfs_id: &str,
mobile_no: &str,
) -> AppResult<Option<EmployeeLookupRow>> {
let cache_key = format!("internal:{}:{}", gtfs_id, mobile_no);
if let Some(cached) = self.get_cached_lookup(&cache_key).await {
return Ok(Some(cached));
}

let internal_pool = match &self.internal_pool {
Some(p) => p,
None => return Ok(None),
};

// Not single-flighted, deterministic on duplicates (see lookup_by_mobile_primary).
let query = r#"
SELECT
e.token_no,
e.first_name,
e.last_name,
e.mobile_no,
e.entity_id,
LOWER(d.designation_name) AS designation_name
FROM employees_internal e
LEFT JOIN designations_internal d ON d.designation_id = e.designation_id
AND d.gtfs_id = e.gtfs_id
AND d.deleted = false
WHERE e.mobile_no = $1
AND e.gtfs_id = $2
AND e.deleted = false
ORDER BY e.emp_id DESC
LIMIT 1
"#;

match sqlx::query_as::<_, EmployeeLookupRow>(query)
.bind(mobile_no)
.bind(gtfs_id)
.fetch_optional(internal_pool)
.await
{
Ok(Some(row)) => {
debug!(
"Employee lookup internal HIT for gtfs_id={} mobile_no={}",
gtfs_id, mobile_no
);
self.cache_lookup(&cache_key, &row).await;
Ok(Some(row))
}
Ok(None) => Ok(None),
Err(e) => Err(AppError::Internal(format!(
"DB query error (internal mobile lookup): {}",
e
))),
}
}
}
Loading