1use std::collections::HashMap;
8use std::sync::Arc;
9
10use axum::Router;
11use serde::{Deserialize, Serialize};
12use uuid::Uuid;
13use chrono::{DateTime, Utc};
14
15use backbone_core::http::{ApiResponse, BackboneCrudHandler};
17
18use backbone_auth::middleware::AuthContext;
20#[cfg(feature = "auth")]
21use backbone_auth::AuthMiddleware;
22
23use crate::domain::entity::*;
25use crate::application::service::{ConversionJobService, ServiceError};
26
27use crate::presentation::dto::{CreateConversionJobDto, UpdateConversionJobDto, PatchConversionJobDto, ConversionJobResponseDto};
29
30use crate::domain::state_machine::{ConversionJobState, ConversionJobStateMachine, ConversionJobTransition};
31
32#[derive(Debug, thiserror::Error)]
34pub enum ConversionJobError {
35 #[error("Not found: {0}")]
36 NotFound(String),
37 #[error("Validation error: {0}")]
38 Validation(String),
39 #[error("Database error: {0}")]
40 Database(String),
41 #[error("Internal error: {0}")]
42 Internal(String),
43 #[error("Source file must exist: {0}")]
45 SourceFileNotFound(String),
46 #[error("Source file must be accessible: {0}")]
47 FileNotAccessible(String),
48 #[error("Target format must be supported: {0}")]
49 ConversionNotSupported(String),
50 #[error("Insufficient storage quota for conversion result: {0}")]
51 InsufficientQuota(String),
52 #[error("A pending conversion with same params already exists: {0}")]
53 DuplicateConversion(String),
54 #[error("Progress cannot decrease: {0}")]
55 InvalidProgress(String),
56}
57
58impl From<ServiceError> for ConversionJobError {
59 fn from(err: ServiceError) -> Self {
60 match err {
61 ServiceError::NotFound => Self::NotFound(err.to_string()),
62 ServiceError::Validation(ref msg) => Self::Validation(msg.clone()),
63 ServiceError::AlreadyExists(ref msg) => Self::Validation(msg.clone()),
64 ServiceError::Repository(ref e) => Self::Database(e.to_string()),
65 ServiceError::Internal(ref msg) => Self::Internal(msg.clone()),
66 ServiceError::Violations(_) => Self::Validation(err.to_string()),
67 }
68 }
69}
70
71impl axum::response::IntoResponse for ConversionJobError {
72 fn into_response(self) -> axum::response::Response {
73 use axum::http::StatusCode;
74 use axum::Json;
75
76 let (status, code) = match &self {
77 Self::NotFound(_) => (StatusCode::NOT_FOUND, "CONVERSIONJOB_NOT_FOUND"),
78 Self::Validation(_) => (StatusCode::BAD_REQUEST, "CONVERSIONJOB_VALIDATION_ERROR"),
79 Self::Database(_) => (StatusCode::INTERNAL_SERVER_ERROR, "CONVERSIONJOB_DATABASE_ERROR"),
80 Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "CONVERSIONJOB_INTERNAL_ERROR"),
81 Self::SourceFileNotFound(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_SOURCE_FILE_NOT_FOUND"),
82 Self::FileNotAccessible(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_FILE_NOT_ACCESSIBLE"),
83 Self::ConversionNotSupported(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_CONVERSION_NOT_SUPPORTED"),
84 Self::InsufficientQuota(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_INSUFFICIENT_QUOTA"),
85 Self::DuplicateConversion(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_DUPLICATE_CONVERSION"),
86 Self::InvalidProgress(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_INVALID_PROGRESS"),
87 };
88
89 let body = serde_json::json!({
90 "success": false,
91 "error": code,
92 "message": self.to_string(),
93 });
94
95 (status, Json(body)).into_response()
96 }
97}
98
99pub mod conversion_job_errors {
101 pub const SOURCE_FILE_NOT_FOUND: &str = "CONVERSIONJOB_SOURCE_FILE_NOT_FOUND";
102 pub const FILE_NOT_ACCESSIBLE: &str = "CONVERSIONJOB_FILE_NOT_ACCESSIBLE";
103 pub const CONVERSION_NOT_SUPPORTED: &str = "CONVERSIONJOB_CONVERSION_NOT_SUPPORTED";
104 pub const INSUFFICIENT_QUOTA: &str = "CONVERSIONJOB_INSUFFICIENT_QUOTA";
105 pub const DUPLICATE_CONVERSION: &str = "CONVERSIONJOB_DUPLICATE_CONVERSION";
106 pub const INVALID_PROGRESS: &str = "CONVERSIONJOB_INVALID_PROGRESS";
107}
108
109pub fn create_conversion_job_routes(service: Arc<ConversionJobService>) -> Router {
142 BackboneCrudHandler::<ConversionJobService, ConversionJob, CreateConversionJobDto, UpdateConversionJobDto, ConversionJobResponseDto>::routes(
143 service,
144 "/conversion_jobs",
145 )
146}
147
148pub fn create_conversion_job_read_routes(service: Arc<ConversionJobService>) -> Router {
154 BackboneCrudHandler::<ConversionJobService, ConversionJob, CreateConversionJobDto, UpdateConversionJobDto, ConversionJobResponseDto>::read_routes(
155 service,
156 "/conversion_jobs",
157 )
158}
159
160pub fn create_conversion_job_write_routes(service: Arc<ConversionJobService>) -> Router {
172 BackboneCrudHandler::<ConversionJobService, ConversionJob, CreateConversionJobDto, UpdateConversionJobDto, ConversionJobResponseDto>::write_routes(
173 service,
174 "/conversion_jobs",
175 )
176}
177
178#[cfg(feature = "auth")]
184pub fn create_protected_conversion_job_routes<A: AuthMiddleware + Send + Sync + 'static>(
185 service: Arc<ConversionJobService>,
186 auth: Arc<A>,
187) -> Router {
188 use axum::middleware;
189 use axum::response::IntoResponse;
190
191 let auth_layer = auth.clone();
192 create_conversion_job_routes(service)
193 .layer(middleware::from_fn(move |mut req: axum::extract::Request, next: axum::middleware::Next| {
194 let auth = auth_layer.clone();
195 async move {
196 let token = req.headers()
197 .get(axum::http::header::AUTHORIZATION)
198 .and_then(|h| h.to_str().ok())
199 .and_then(|raw| raw.strip_prefix("Bearer ").or_else(|| raw.strip_prefix("bearer ")))
200 .unwrap_or("");
201 match auth.authenticate(token).await {
202 Ok(ctx) => {
203 req.extensions_mut().insert(ctx);
204 next.run(req).await
205 }
206 Err(_) => {
207 (axum::http::StatusCode::UNAUTHORIZED,
208 axum::Json(serde_json::json!({
209 "success": false,
210 "error": "unauthorized",
211 "message": "Authentication required"
212 }))
213 ).into_response()
214 }
215 }
216 }
217 }))
218}
219
220pub async fn start_transition(
228 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
229 axum::extract::Path(id): axum::extract::Path<String>,
230 auth: Option<axum::Extension<AuthContext>>,
231) -> impl axum::response::IntoResponse {
232 use axum::{http::StatusCode, Json};
233
234 let entity = match service.get_by_id(&id).await {
236 Ok(Some(e)) => e,
237 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
238 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
239 };
240
241 let Some(axum::Extension(auth)) = auth else {
243 return (StatusCode::UNAUTHORIZED, Json(ApiResponse::<ConversionJobResponseDto>::error("A transition needs an authenticated caller")));
244 };
245 {
246 let allowed_roles = ConversionJobTransition::Start.allowed_roles();
247 let granted = |required: &str| auth.permissions.iter().any(|p| {
248 let (have, want): (Vec<&str>, Vec<&str>) = (p.split(':').collect(), required.split(':').collect());
249 have.iter().enumerate().all(|(i, h)| *h == "*" || want.get(i) == Some(h))
250 && (have.len() == want.len() || (have.len() < want.len() && have.last() == Some(&"*")))
251 });
252 let has_specific_perm = granted("conversion_job:transition:start");
253 let has_update_perm = granted("conversion_job:update");
254 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
255 if !has_specific_perm && !has_role {
256 if !has_update_perm {
257 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for start transition")));
258 }
259 tracing::warn!(target: "backbone::write_guard", entity = "ConversionJob", verb = "start", code = "transition_by_update_permission", "a transition granted only by conversion_job:update, running in shadow; it needs conversion_job:transition:start or one of the roles system");
260 }
261 }
262
263 let current_state: ConversionJobState = match entity.status.to_string().parse() {
265 Ok(s) => s,
266 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
267 };
268 let sm = ConversionJobStateMachine::from_state(current_state);
269 if !sm.can_transition(ConversionJobTransition::Start) {
270 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
271 }
272
273 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
276 fields.insert("status".to_string(), serde_json::Value::String("processing".to_string()));
277
278 match backbone_core::write_guard::as_transition("start", service.partial_update(&id, fields)).await {
279 Ok(Some(updated)) => {
280 let response: ConversionJobResponseDto = updated.into();
281 (StatusCode::OK, Json(ApiResponse::ok(response)))
282 }
283 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
284 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
285 }
286}
287
288pub async fn update_progress_transition(
292 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
293 axum::extract::Path(id): axum::extract::Path<String>,
294 auth: Option<axum::Extension<AuthContext>>,
295) -> impl axum::response::IntoResponse {
296 use axum::{http::StatusCode, Json};
297
298 let entity = match service.get_by_id(&id).await {
300 Ok(Some(e)) => e,
301 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
302 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
303 };
304
305 let Some(axum::Extension(auth)) = auth else {
307 return (StatusCode::UNAUTHORIZED, Json(ApiResponse::<ConversionJobResponseDto>::error("A transition needs an authenticated caller")));
308 };
309 {
310 let allowed_roles = ConversionJobTransition::UpdateProgress.allowed_roles();
311 let granted = |required: &str| auth.permissions.iter().any(|p| {
312 let (have, want): (Vec<&str>, Vec<&str>) = (p.split(':').collect(), required.split(':').collect());
313 have.iter().enumerate().all(|(i, h)| *h == "*" || want.get(i) == Some(h))
314 && (have.len() == want.len() || (have.len() < want.len() && have.last() == Some(&"*")))
315 });
316 let has_specific_perm = granted("conversion_job:transition:update_progress");
317 let has_update_perm = granted("conversion_job:update");
318 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
319 if !has_specific_perm && !has_role {
320 if !has_update_perm {
321 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for update_progress transition")));
322 }
323 tracing::warn!(target: "backbone::write_guard", entity = "ConversionJob", verb = "update_progress", code = "transition_by_update_permission", "a transition granted only by conversion_job:update, running in shadow; it needs conversion_job:transition:update_progress or one of the roles system");
324 }
325 }
326
327 let current_state: ConversionJobState = match entity.status.to_string().parse() {
329 Ok(s) => s,
330 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
331 };
332 let sm = ConversionJobStateMachine::from_state(current_state);
333 if !sm.can_transition(ConversionJobTransition::UpdateProgress) {
334 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
335 }
336
337 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
340 fields.insert("status".to_string(), serde_json::Value::String("processing".to_string()));
341
342 match backbone_core::write_guard::as_transition("update_progress", service.partial_update(&id, fields)).await {
343 Ok(Some(updated)) => {
344 let response: ConversionJobResponseDto = updated.into();
345 (StatusCode::OK, Json(ApiResponse::ok(response)))
346 }
347 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
348 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
349 }
350}
351
352pub async fn complete_transition(
356 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
357 axum::extract::Path(id): axum::extract::Path<String>,
358 auth: Option<axum::Extension<AuthContext>>,
359) -> impl axum::response::IntoResponse {
360 use axum::{http::StatusCode, Json};
361
362 let entity = match service.get_by_id(&id).await {
364 Ok(Some(e)) => e,
365 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
366 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
367 };
368
369 let Some(axum::Extension(auth)) = auth else {
371 return (StatusCode::UNAUTHORIZED, Json(ApiResponse::<ConversionJobResponseDto>::error("A transition needs an authenticated caller")));
372 };
373 {
374 let allowed_roles = ConversionJobTransition::Complete.allowed_roles();
375 let granted = |required: &str| auth.permissions.iter().any(|p| {
376 let (have, want): (Vec<&str>, Vec<&str>) = (p.split(':').collect(), required.split(':').collect());
377 have.iter().enumerate().all(|(i, h)| *h == "*" || want.get(i) == Some(h))
378 && (have.len() == want.len() || (have.len() < want.len() && have.last() == Some(&"*")))
379 });
380 let has_specific_perm = granted("conversion_job:transition:complete");
381 let has_update_perm = granted("conversion_job:update");
382 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
383 if !has_specific_perm && !has_role {
384 if !has_update_perm {
385 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for complete transition")));
386 }
387 tracing::warn!(target: "backbone::write_guard", entity = "ConversionJob", verb = "complete", code = "transition_by_update_permission", "a transition granted only by conversion_job:update, running in shadow; it needs conversion_job:transition:complete or one of the roles system");
388 }
389 }
390
391 let current_state: ConversionJobState = match entity.status.to_string().parse() {
393 Ok(s) => s,
394 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
395 };
396 let sm = ConversionJobStateMachine::from_state(current_state);
397 if !sm.can_transition(ConversionJobTransition::Complete) {
398 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
399 }
400
401 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
403 fields.insert("status".to_string(), serde_json::Value::String("completed".to_string()));
404
405 match backbone_core::write_guard::as_transition("complete", service.partial_update(&id, fields)).await {
406 Ok(Some(updated)) => {
407 let response: ConversionJobResponseDto = updated.into();
408 (StatusCode::OK, Json(ApiResponse::ok(response)))
409 }
410 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
411 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
412 }
413}
414
415pub async fn fail_transition(
419 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
420 axum::extract::Path(id): axum::extract::Path<String>,
421 auth: Option<axum::Extension<AuthContext>>,
422) -> impl axum::response::IntoResponse {
423 use axum::{http::StatusCode, Json};
424
425 let entity = match service.get_by_id(&id).await {
427 Ok(Some(e)) => e,
428 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
429 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
430 };
431
432 let Some(axum::Extension(auth)) = auth else {
434 return (StatusCode::UNAUTHORIZED, Json(ApiResponse::<ConversionJobResponseDto>::error("A transition needs an authenticated caller")));
435 };
436 {
437 let allowed_roles = ConversionJobTransition::Fail.allowed_roles();
438 let granted = |required: &str| auth.permissions.iter().any(|p| {
439 let (have, want): (Vec<&str>, Vec<&str>) = (p.split(':').collect(), required.split(':').collect());
440 have.iter().enumerate().all(|(i, h)| *h == "*" || want.get(i) == Some(h))
441 && (have.len() == want.len() || (have.len() < want.len() && have.last() == Some(&"*")))
442 });
443 let has_specific_perm = granted("conversion_job:transition:fail");
444 let has_update_perm = granted("conversion_job:update");
445 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
446 if !has_specific_perm && !has_role {
447 if !has_update_perm {
448 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for fail transition")));
449 }
450 tracing::warn!(target: "backbone::write_guard", entity = "ConversionJob", verb = "fail", code = "transition_by_update_permission", "a transition granted only by conversion_job:update, running in shadow; it needs conversion_job:transition:fail or one of the roles system");
451 }
452 }
453
454 let current_state: ConversionJobState = match entity.status.to_string().parse() {
456 Ok(s) => s,
457 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
458 };
459 let sm = ConversionJobStateMachine::from_state(current_state);
460 if !sm.can_transition(ConversionJobTransition::Fail) {
461 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
462 }
463
464 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
466 fields.insert("status".to_string(), serde_json::Value::String("failed".to_string()));
467
468 match backbone_core::write_guard::as_transition("fail", service.partial_update(&id, fields)).await {
469 Ok(Some(updated)) => {
470 let response: ConversionJobResponseDto = updated.into();
471 (StatusCode::OK, Json(ApiResponse::ok(response)))
472 }
473 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
474 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
475 }
476}
477
478pub async fn cancel_pending_transition(
482 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
483 axum::extract::Path(id): axum::extract::Path<String>,
484 auth: Option<axum::Extension<AuthContext>>,
485) -> impl axum::response::IntoResponse {
486 use axum::{http::StatusCode, Json};
487
488 let entity = match service.get_by_id(&id).await {
490 Ok(Some(e)) => e,
491 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
492 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
493 };
494
495 let Some(axum::Extension(auth)) = auth else {
497 return (StatusCode::UNAUTHORIZED, Json(ApiResponse::<ConversionJobResponseDto>::error("A transition needs an authenticated caller")));
498 };
499 {
500 let allowed_roles = ConversionJobTransition::CancelPending.allowed_roles();
501 let granted = |required: &str| auth.permissions.iter().any(|p| {
502 let (have, want): (Vec<&str>, Vec<&str>) = (p.split(':').collect(), required.split(':').collect());
503 have.iter().enumerate().all(|(i, h)| *h == "*" || want.get(i) == Some(h))
504 && (have.len() == want.len() || (have.len() < want.len() && have.last() == Some(&"*")))
505 });
506 let has_specific_perm = granted("conversion_job:transition:cancel_pending");
507 let has_update_perm = granted("conversion_job:update");
508 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
509 if !has_specific_perm && !has_role {
510 if !has_update_perm {
511 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for cancel_pending transition")));
512 }
513 tracing::warn!(target: "backbone::write_guard", entity = "ConversionJob", verb = "cancel_pending", code = "transition_by_update_permission", "a transition granted only by conversion_job:update, running in shadow; it needs conversion_job:transition:cancel_pending or one of the roles owner, admin");
514 }
515 }
516
517 let current_state: ConversionJobState = match entity.status.to_string().parse() {
519 Ok(s) => s,
520 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
521 };
522 let sm = ConversionJobStateMachine::from_state(current_state);
523 if !sm.can_transition(ConversionJobTransition::CancelPending) {
524 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
525 }
526
527 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
530 fields.insert("status".to_string(), serde_json::Value::String("cancelled".to_string()));
531
532 match backbone_core::write_guard::as_transition("cancel_pending", service.partial_update(&id, fields)).await {
533 Ok(Some(updated)) => {
534 let response: ConversionJobResponseDto = updated.into();
535 (StatusCode::OK, Json(ApiResponse::ok(response)))
536 }
537 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
538 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
539 }
540}
541
542pub async fn cancel_running_transition(
546 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
547 axum::extract::Path(id): axum::extract::Path<String>,
548 auth: Option<axum::Extension<AuthContext>>,
549) -> impl axum::response::IntoResponse {
550 use axum::{http::StatusCode, Json};
551
552 let entity = match service.get_by_id(&id).await {
554 Ok(Some(e)) => e,
555 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
556 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
557 };
558
559 let Some(axum::Extension(auth)) = auth else {
561 return (StatusCode::UNAUTHORIZED, Json(ApiResponse::<ConversionJobResponseDto>::error("A transition needs an authenticated caller")));
562 };
563 {
564 let allowed_roles = ConversionJobTransition::CancelRunning.allowed_roles();
565 let granted = |required: &str| auth.permissions.iter().any(|p| {
566 let (have, want): (Vec<&str>, Vec<&str>) = (p.split(':').collect(), required.split(':').collect());
567 have.iter().enumerate().all(|(i, h)| *h == "*" || want.get(i) == Some(h))
568 && (have.len() == want.len() || (have.len() < want.len() && have.last() == Some(&"*")))
569 });
570 let has_specific_perm = granted("conversion_job:transition:cancel_running");
571 let has_update_perm = granted("conversion_job:update");
572 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
573 if !has_specific_perm && !has_role {
574 if !has_update_perm {
575 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for cancel_running transition")));
576 }
577 tracing::warn!(target: "backbone::write_guard", entity = "ConversionJob", verb = "cancel_running", code = "transition_by_update_permission", "a transition granted only by conversion_job:update, running in shadow; it needs conversion_job:transition:cancel_running or one of the roles owner, admin");
578 }
579 }
580
581 let current_state: ConversionJobState = match entity.status.to_string().parse() {
583 Ok(s) => s,
584 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
585 };
586 let sm = ConversionJobStateMachine::from_state(current_state);
587 if !sm.can_transition(ConversionJobTransition::CancelRunning) {
588 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
589 }
590
591 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
594 fields.insert("status".to_string(), serde_json::Value::String("cancelled".to_string()));
595
596 match backbone_core::write_guard::as_transition("cancel_running", service.partial_update(&id, fields)).await {
597 Ok(Some(updated)) => {
598 let response: ConversionJobResponseDto = updated.into();
599 (StatusCode::OK, Json(ApiResponse::ok(response)))
600 }
601 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
602 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
603 }
604}
605
606pub async fn retry_transition(
610 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
611 axum::extract::Path(id): axum::extract::Path<String>,
612 auth: Option<axum::Extension<AuthContext>>,
613) -> impl axum::response::IntoResponse {
614 use axum::{http::StatusCode, Json};
615
616 let entity = match service.get_by_id(&id).await {
618 Ok(Some(e)) => e,
619 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
620 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
621 };
622
623 let Some(axum::Extension(auth)) = auth else {
625 return (StatusCode::UNAUTHORIZED, Json(ApiResponse::<ConversionJobResponseDto>::error("A transition needs an authenticated caller")));
626 };
627 {
628 let allowed_roles = ConversionJobTransition::Retry.allowed_roles();
629 let granted = |required: &str| auth.permissions.iter().any(|p| {
630 let (have, want): (Vec<&str>, Vec<&str>) = (p.split(':').collect(), required.split(':').collect());
631 have.iter().enumerate().all(|(i, h)| *h == "*" || want.get(i) == Some(h))
632 && (have.len() == want.len() || (have.len() < want.len() && have.last() == Some(&"*")))
633 });
634 let has_specific_perm = granted("conversion_job:transition:retry");
635 let has_update_perm = granted("conversion_job:update");
636 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
637 if !has_specific_perm && !has_role {
638 if !has_update_perm {
639 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for retry transition")));
640 }
641 tracing::warn!(target: "backbone::write_guard", entity = "ConversionJob", verb = "retry", code = "transition_by_update_permission", "a transition granted only by conversion_job:update, running in shadow; it needs conversion_job:transition:retry or one of the roles system, owner");
642 }
643 }
644
645 let current_state: ConversionJobState = match entity.status.to_string().parse() {
647 Ok(s) => s,
648 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
649 };
650 let sm = ConversionJobStateMachine::from_state(current_state);
651 if !sm.can_transition(ConversionJobTransition::Retry) {
652 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
653 }
654
655 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
658 fields.insert("status".to_string(), serde_json::Value::String("pending".to_string()));
659
660 match backbone_core::write_guard::as_transition("retry", service.partial_update(&id, fields)).await {
661 Ok(Some(updated)) => {
662 let response: ConversionJobResponseDto = updated.into();
663 (StatusCode::OK, Json(ApiResponse::ok(response)))
664 }
665 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
666 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
667 }
668}
669
670pub async fn reconvert_transition(
674 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
675 axum::extract::Path(id): axum::extract::Path<String>,
676 auth: Option<axum::Extension<AuthContext>>,
677) -> impl axum::response::IntoResponse {
678 use axum::{http::StatusCode, Json};
679
680 let entity = match service.get_by_id(&id).await {
682 Ok(Some(e)) => e,
683 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
684 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
685 };
686
687 let Some(axum::Extension(auth)) = auth else {
689 return (StatusCode::UNAUTHORIZED, Json(ApiResponse::<ConversionJobResponseDto>::error("A transition needs an authenticated caller")));
690 };
691 {
692 let allowed_roles = ConversionJobTransition::Reconvert.allowed_roles();
693 let granted = |required: &str| auth.permissions.iter().any(|p| {
694 let (have, want): (Vec<&str>, Vec<&str>) = (p.split(':').collect(), required.split(':').collect());
695 have.iter().enumerate().all(|(i, h)| *h == "*" || want.get(i) == Some(h))
696 && (have.len() == want.len() || (have.len() < want.len() && have.last() == Some(&"*")))
697 });
698 let has_specific_perm = granted("conversion_job:transition:reconvert");
699 let has_update_perm = granted("conversion_job:update");
700 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
701 if !has_specific_perm && !has_role {
702 if !has_update_perm {
703 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for reconvert transition")));
704 }
705 tracing::warn!(target: "backbone::write_guard", entity = "ConversionJob", verb = "reconvert", code = "transition_by_update_permission", "a transition granted only by conversion_job:update, running in shadow; it needs conversion_job:transition:reconvert or one of the roles owner, admin");
706 }
707 }
708
709 let current_state: ConversionJobState = match entity.status.to_string().parse() {
711 Ok(s) => s,
712 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
713 };
714 let sm = ConversionJobStateMachine::from_state(current_state);
715 if !sm.can_transition(ConversionJobTransition::Reconvert) {
716 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
717 }
718
719 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
721 fields.insert("status".to_string(), serde_json::Value::String("pending".to_string()));
722
723 match backbone_core::write_guard::as_transition("reconvert", service.partial_update(&id, fields)).await {
724 Ok(Some(updated)) => {
725 let response: ConversionJobResponseDto = updated.into();
726 (StatusCode::OK, Json(ApiResponse::ok(response)))
727 }
728 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
729 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
730 }
731}
732
733pub fn create_conversion_job_transition_routes(service: Arc<ConversionJobService>) -> Router {
735 use axum::routing::post;
736
737 Router::new()
738 .route("/conversion_jobs/:id/transitions/start", post(start_transition))
739 .route("/conversion_jobs/:id/transitions/update_progress", post(update_progress_transition))
740 .route("/conversion_jobs/:id/transitions/complete", post(complete_transition))
741 .route("/conversion_jobs/:id/transitions/fail", post(fail_transition))
742 .route("/conversion_jobs/:id/transitions/cancel_pending", post(cancel_pending_transition))
743 .route("/conversion_jobs/:id/transitions/cancel_running", post(cancel_running_transition))
744 .route("/conversion_jobs/:id/transitions/retry", post(retry_transition))
745 .route("/conversion_jobs/:id/transitions/reconvert", post(reconvert_transition))
746 .with_state(service)
747}
748