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
18#[cfg(feature = "auth")]
20use backbone_auth::middleware::AuthContext;
21#[cfg(feature = "auth")]
22use backbone_auth::AuthMiddleware;
23
24use crate::domain::entity::*;
26use crate::application::service::{ConversionJobService, ServiceError};
27
28use crate::presentation::dto::{CreateConversionJobDto, UpdateConversionJobDto, PatchConversionJobDto, ConversionJobResponseDto};
30
31use crate::domain::state_machine::{ConversionJobState, ConversionJobStateMachine, ConversionJobTransition};
32
33#[derive(Debug, thiserror::Error)]
35pub enum ConversionJobError {
36 #[error("Not found: {0}")]
37 NotFound(String),
38 #[error("Validation error: {0}")]
39 Validation(String),
40 #[error("Database error: {0}")]
41 Database(String),
42 #[error("Internal error: {0}")]
43 Internal(String),
44 #[error("Source file must exist: {0}")]
46 SourceFileNotFound(String),
47 #[error("Source file must be accessible: {0}")]
48 FileNotAccessible(String),
49 #[error("Target format must be supported: {0}")]
50 ConversionNotSupported(String),
51 #[error("Insufficient storage quota for conversion result: {0}")]
52 InsufficientQuota(String),
53 #[error("A pending conversion with same params already exists: {0}")]
54 DuplicateConversion(String),
55 #[error("Progress cannot decrease: {0}")]
56 InvalidProgress(String),
57}
58
59impl From<ServiceError> for ConversionJobError {
60 fn from(err: ServiceError) -> Self {
61 match err {
62 ServiceError::NotFound => Self::NotFound(err.to_string()),
63 ServiceError::Validation(ref msg) => Self::Validation(msg.clone()),
64 ServiceError::AlreadyExists(ref msg) => Self::Validation(msg.clone()),
65 ServiceError::Repository(ref e) => Self::Database(e.to_string()),
66 ServiceError::Internal(ref msg) => Self::Internal(msg.clone()),
67 ServiceError::Violations(_) => Self::Validation(err.to_string()),
68 }
69 }
70}
71
72impl axum::response::IntoResponse for ConversionJobError {
73 fn into_response(self) -> axum::response::Response {
74 use axum::http::StatusCode;
75 use axum::Json;
76
77 let (status, code) = match &self {
78 Self::NotFound(_) => (StatusCode::NOT_FOUND, "CONVERSIONJOB_NOT_FOUND"),
79 Self::Validation(_) => (StatusCode::BAD_REQUEST, "CONVERSIONJOB_VALIDATION_ERROR"),
80 Self::Database(_) => (StatusCode::INTERNAL_SERVER_ERROR, "CONVERSIONJOB_DATABASE_ERROR"),
81 Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "CONVERSIONJOB_INTERNAL_ERROR"),
82 Self::SourceFileNotFound(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_SOURCE_FILE_NOT_FOUND"),
83 Self::FileNotAccessible(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_FILE_NOT_ACCESSIBLE"),
84 Self::ConversionNotSupported(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_CONVERSION_NOT_SUPPORTED"),
85 Self::InsufficientQuota(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_INSUFFICIENT_QUOTA"),
86 Self::DuplicateConversion(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_DUPLICATE_CONVERSION"),
87 Self::InvalidProgress(_) => (StatusCode::UNPROCESSABLE_ENTITY, "CONVERSIONJOB_INVALID_PROGRESS"),
88 };
89
90 let body = serde_json::json!({
91 "success": false,
92 "error": code,
93 "message": self.to_string(),
94 });
95
96 (status, Json(body)).into_response()
97 }
98}
99
100pub mod conversion_job_errors {
102 pub const SOURCE_FILE_NOT_FOUND: &str = "CONVERSIONJOB_SOURCE_FILE_NOT_FOUND";
103 pub const FILE_NOT_ACCESSIBLE: &str = "CONVERSIONJOB_FILE_NOT_ACCESSIBLE";
104 pub const CONVERSION_NOT_SUPPORTED: &str = "CONVERSIONJOB_CONVERSION_NOT_SUPPORTED";
105 pub const INSUFFICIENT_QUOTA: &str = "CONVERSIONJOB_INSUFFICIENT_QUOTA";
106 pub const DUPLICATE_CONVERSION: &str = "CONVERSIONJOB_DUPLICATE_CONVERSION";
107 pub const INVALID_PROGRESS: &str = "CONVERSIONJOB_INVALID_PROGRESS";
108}
109
110pub fn create_conversion_job_routes(service: Arc<ConversionJobService>) -> Router {
143 BackboneCrudHandler::<ConversionJobService, ConversionJob, CreateConversionJobDto, UpdateConversionJobDto, ConversionJobResponseDto>::routes(
144 service,
145 "/conversion_jobs",
146 )
147}
148
149pub fn create_conversion_job_read_routes(service: Arc<ConversionJobService>) -> Router {
155 BackboneCrudHandler::<ConversionJobService, ConversionJob, CreateConversionJobDto, UpdateConversionJobDto, ConversionJobResponseDto>::read_routes(
156 service,
157 "/conversion_jobs",
158 )
159}
160
161pub fn create_conversion_job_write_routes(service: Arc<ConversionJobService>) -> Router {
173 BackboneCrudHandler::<ConversionJobService, ConversionJob, CreateConversionJobDto, UpdateConversionJobDto, ConversionJobResponseDto>::write_routes(
174 service,
175 "/conversion_jobs",
176 )
177}
178
179#[cfg(feature = "auth")]
185pub fn create_protected_conversion_job_routes<A: AuthMiddleware + Send + Sync + 'static>(
186 service: Arc<ConversionJobService>,
187 auth: Arc<A>,
188) -> Router {
189 use axum::middleware;
190 use axum::response::IntoResponse;
191
192 let auth_layer = auth.clone();
193 create_conversion_job_routes(service)
194 .layer(middleware::from_fn(move |mut req: axum::extract::Request, next: axum::middleware::Next| {
195 let auth = auth_layer.clone();
196 async move {
197 let token = req.headers()
198 .get(axum::http::header::AUTHORIZATION)
199 .and_then(|h| h.to_str().ok())
200 .and_then(|raw| raw.strip_prefix("Bearer ").or_else(|| raw.strip_prefix("bearer ")))
201 .unwrap_or("");
202 match auth.authenticate(token).await {
203 Ok(ctx) => {
204 req.extensions_mut().insert(ctx);
205 next.run(req).await
206 }
207 Err(_) => {
208 (axum::http::StatusCode::UNAUTHORIZED,
209 axum::Json(serde_json::json!({
210 "success": false,
211 "error": "unauthorized",
212 "message": "Authentication required"
213 }))
214 ).into_response()
215 }
216 }
217 }
218 }))
219}
220
221pub async fn start_transition(
229 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
230 axum::extract::Path(id): axum::extract::Path<String>,
231 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
232) -> impl axum::response::IntoResponse {
233 use axum::{http::StatusCode, Json};
234
235 let entity = match service.get_by_id(&id).await {
237 Ok(Some(e)) => e,
238 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
239 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
240 };
241
242 #[cfg(feature = "auth")]
244 {
245 let allowed_roles = ConversionJobTransition::Start.allowed_roles();
246 let has_specific_perm = auth.permissions.iter().any(|p| p == "conversion_job:transition:start");
247 let has_update_perm = auth.permissions.iter().any(|p| p == "conversion_job:update");
248 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
249 if !has_specific_perm && !has_role {
250 if !has_update_perm {
251 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for start transition")));
252 }
253 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 a role of ["system"]");
254 }
255 }
256
257 let current_state: ConversionJobState = match entity.status.to_string().parse() {
259 Ok(s) => s,
260 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
261 };
262 let sm = ConversionJobStateMachine::from_state(current_state);
263 if !sm.can_transition(ConversionJobTransition::Start) {
264 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
265 }
266
267 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
270 fields.insert("status".to_string(), serde_json::Value::String("processing".to_string()));
271
272 match backbone_core::write_guard::as_transition("start", service.partial_update(&id, fields)).await {
273 Ok(Some(updated)) => {
274 let response: ConversionJobResponseDto = updated.into();
275 (StatusCode::OK, Json(ApiResponse::ok(response)))
276 }
277 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
278 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
279 }
280}
281
282pub async fn update_progress_transition(
286 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
287 axum::extract::Path(id): axum::extract::Path<String>,
288 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
289) -> impl axum::response::IntoResponse {
290 use axum::{http::StatusCode, Json};
291
292 let entity = match service.get_by_id(&id).await {
294 Ok(Some(e)) => e,
295 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
296 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
297 };
298
299 #[cfg(feature = "auth")]
301 {
302 let allowed_roles = ConversionJobTransition::UpdateProgress.allowed_roles();
303 let has_specific_perm = auth.permissions.iter().any(|p| p == "conversion_job:transition:update_progress");
304 let has_update_perm = auth.permissions.iter().any(|p| p == "conversion_job:update");
305 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
306 if !has_specific_perm && !has_role {
307 if !has_update_perm {
308 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for update_progress transition")));
309 }
310 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 a role of ["system"]");
311 }
312 }
313
314 let current_state: ConversionJobState = match entity.status.to_string().parse() {
316 Ok(s) => s,
317 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
318 };
319 let sm = ConversionJobStateMachine::from_state(current_state);
320 if !sm.can_transition(ConversionJobTransition::UpdateProgress) {
321 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
322 }
323
324 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
327 fields.insert("status".to_string(), serde_json::Value::String("processing".to_string()));
328
329 match backbone_core::write_guard::as_transition("update_progress", service.partial_update(&id, fields)).await {
330 Ok(Some(updated)) => {
331 let response: ConversionJobResponseDto = updated.into();
332 (StatusCode::OK, Json(ApiResponse::ok(response)))
333 }
334 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
335 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
336 }
337}
338
339pub async fn complete_transition(
343 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
344 axum::extract::Path(id): axum::extract::Path<String>,
345 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
346) -> impl axum::response::IntoResponse {
347 use axum::{http::StatusCode, Json};
348
349 let entity = match service.get_by_id(&id).await {
351 Ok(Some(e)) => e,
352 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
353 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
354 };
355
356 #[cfg(feature = "auth")]
358 {
359 let allowed_roles = ConversionJobTransition::Complete.allowed_roles();
360 let has_specific_perm = auth.permissions.iter().any(|p| p == "conversion_job:transition:complete");
361 let has_update_perm = auth.permissions.iter().any(|p| p == "conversion_job:update");
362 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
363 if !has_specific_perm && !has_role {
364 if !has_update_perm {
365 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for complete transition")));
366 }
367 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 a role of ["system"]");
368 }
369 }
370
371 let current_state: ConversionJobState = match entity.status.to_string().parse() {
373 Ok(s) => s,
374 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
375 };
376 let sm = ConversionJobStateMachine::from_state(current_state);
377 if !sm.can_transition(ConversionJobTransition::Complete) {
378 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
379 }
380
381 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
383 fields.insert("status".to_string(), serde_json::Value::String("completed".to_string()));
384
385 match backbone_core::write_guard::as_transition("complete", service.partial_update(&id, fields)).await {
386 Ok(Some(updated)) => {
387 let response: ConversionJobResponseDto = updated.into();
388 (StatusCode::OK, Json(ApiResponse::ok(response)))
389 }
390 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
391 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
392 }
393}
394
395pub async fn fail_transition(
399 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
400 axum::extract::Path(id): axum::extract::Path<String>,
401 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
402) -> impl axum::response::IntoResponse {
403 use axum::{http::StatusCode, Json};
404
405 let entity = match service.get_by_id(&id).await {
407 Ok(Some(e)) => e,
408 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
409 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
410 };
411
412 #[cfg(feature = "auth")]
414 {
415 let allowed_roles = ConversionJobTransition::Fail.allowed_roles();
416 let has_specific_perm = auth.permissions.iter().any(|p| p == "conversion_job:transition:fail");
417 let has_update_perm = auth.permissions.iter().any(|p| p == "conversion_job:update");
418 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
419 if !has_specific_perm && !has_role {
420 if !has_update_perm {
421 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for fail transition")));
422 }
423 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 a role of ["system"]");
424 }
425 }
426
427 let current_state: ConversionJobState = match entity.status.to_string().parse() {
429 Ok(s) => s,
430 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
431 };
432 let sm = ConversionJobStateMachine::from_state(current_state);
433 if !sm.can_transition(ConversionJobTransition::Fail) {
434 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
435 }
436
437 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
439 fields.insert("status".to_string(), serde_json::Value::String("failed".to_string()));
440
441 match backbone_core::write_guard::as_transition("fail", service.partial_update(&id, fields)).await {
442 Ok(Some(updated)) => {
443 let response: ConversionJobResponseDto = updated.into();
444 (StatusCode::OK, Json(ApiResponse::ok(response)))
445 }
446 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
447 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
448 }
449}
450
451pub async fn cancel_pending_transition(
455 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
456 axum::extract::Path(id): axum::extract::Path<String>,
457 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
458) -> impl axum::response::IntoResponse {
459 use axum::{http::StatusCode, Json};
460
461 let entity = match service.get_by_id(&id).await {
463 Ok(Some(e)) => e,
464 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
465 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
466 };
467
468 #[cfg(feature = "auth")]
470 {
471 let allowed_roles = ConversionJobTransition::CancelPending.allowed_roles();
472 let has_specific_perm = auth.permissions.iter().any(|p| p == "conversion_job:transition:cancel_pending");
473 let has_update_perm = auth.permissions.iter().any(|p| p == "conversion_job:update");
474 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
475 if !has_specific_perm && !has_role {
476 if !has_update_perm {
477 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for cancel_pending transition")));
478 }
479 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 a role of ["owner", "admin"]");
480 }
481 }
482
483 let current_state: ConversionJobState = match entity.status.to_string().parse() {
485 Ok(s) => s,
486 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
487 };
488 let sm = ConversionJobStateMachine::from_state(current_state);
489 if !sm.can_transition(ConversionJobTransition::CancelPending) {
490 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
491 }
492
493 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
496 fields.insert("status".to_string(), serde_json::Value::String("cancelled".to_string()));
497
498 match backbone_core::write_guard::as_transition("cancel_pending", service.partial_update(&id, fields)).await {
499 Ok(Some(updated)) => {
500 let response: ConversionJobResponseDto = updated.into();
501 (StatusCode::OK, Json(ApiResponse::ok(response)))
502 }
503 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
504 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
505 }
506}
507
508pub async fn cancel_running_transition(
512 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
513 axum::extract::Path(id): axum::extract::Path<String>,
514 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
515) -> impl axum::response::IntoResponse {
516 use axum::{http::StatusCode, Json};
517
518 let entity = match service.get_by_id(&id).await {
520 Ok(Some(e)) => e,
521 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
522 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
523 };
524
525 #[cfg(feature = "auth")]
527 {
528 let allowed_roles = ConversionJobTransition::CancelRunning.allowed_roles();
529 let has_specific_perm = auth.permissions.iter().any(|p| p == "conversion_job:transition:cancel_running");
530 let has_update_perm = auth.permissions.iter().any(|p| p == "conversion_job:update");
531 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
532 if !has_specific_perm && !has_role {
533 if !has_update_perm {
534 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for cancel_running transition")));
535 }
536 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 a role of ["owner", "admin"]");
537 }
538 }
539
540 let current_state: ConversionJobState = match entity.status.to_string().parse() {
542 Ok(s) => s,
543 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
544 };
545 let sm = ConversionJobStateMachine::from_state(current_state);
546 if !sm.can_transition(ConversionJobTransition::CancelRunning) {
547 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
548 }
549
550 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
553 fields.insert("status".to_string(), serde_json::Value::String("cancelled".to_string()));
554
555 match backbone_core::write_guard::as_transition("cancel_running", service.partial_update(&id, fields)).await {
556 Ok(Some(updated)) => {
557 let response: ConversionJobResponseDto = updated.into();
558 (StatusCode::OK, Json(ApiResponse::ok(response)))
559 }
560 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
561 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
562 }
563}
564
565pub async fn retry_transition(
569 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
570 axum::extract::Path(id): axum::extract::Path<String>,
571 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
572) -> impl axum::response::IntoResponse {
573 use axum::{http::StatusCode, Json};
574
575 let entity = match service.get_by_id(&id).await {
577 Ok(Some(e)) => e,
578 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
579 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
580 };
581
582 #[cfg(feature = "auth")]
584 {
585 let allowed_roles = ConversionJobTransition::Retry.allowed_roles();
586 let has_specific_perm = auth.permissions.iter().any(|p| p == "conversion_job:transition:retry");
587 let has_update_perm = auth.permissions.iter().any(|p| p == "conversion_job:update");
588 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
589 if !has_specific_perm && !has_role {
590 if !has_update_perm {
591 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for retry transition")));
592 }
593 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 a role of ["system", "owner"]");
594 }
595 }
596
597 let current_state: ConversionJobState = match entity.status.to_string().parse() {
599 Ok(s) => s,
600 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
601 };
602 let sm = ConversionJobStateMachine::from_state(current_state);
603 if !sm.can_transition(ConversionJobTransition::Retry) {
604 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
605 }
606
607 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
610 fields.insert("status".to_string(), serde_json::Value::String("pending".to_string()));
611
612 match backbone_core::write_guard::as_transition("retry", service.partial_update(&id, fields)).await {
613 Ok(Some(updated)) => {
614 let response: ConversionJobResponseDto = updated.into();
615 (StatusCode::OK, Json(ApiResponse::ok(response)))
616 }
617 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
618 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
619 }
620}
621
622pub async fn reconvert_transition(
626 axum::extract::State(service): axum::extract::State<Arc<ConversionJobService>>,
627 axum::extract::Path(id): axum::extract::Path<String>,
628 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
629) -> impl axum::response::IntoResponse {
630 use axum::{http::StatusCode, Json};
631
632 let entity = match service.get_by_id(&id).await {
634 Ok(Some(e)) => e,
635 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
636 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
637 };
638
639 #[cfg(feature = "auth")]
641 {
642 let allowed_roles = ConversionJobTransition::Reconvert.allowed_roles();
643 let has_specific_perm = auth.permissions.iter().any(|p| p == "conversion_job:transition:reconvert");
644 let has_update_perm = auth.permissions.iter().any(|p| p == "conversion_job:update");
645 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
646 if !has_specific_perm && !has_role {
647 if !has_update_perm {
648 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ConversionJobResponseDto>::error("Insufficient permissions for reconvert transition")));
649 }
650 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 a role of ["owner", "admin"]");
651 }
652 }
653
654 let current_state: ConversionJobState = match entity.status.to_string().parse() {
656 Ok(s) => s,
657 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ConversionJobResponseDto>::error("The stored state is not a state of this machine"))),
658 };
659 let sm = ConversionJobStateMachine::from_state(current_state);
660 if !sm.can_transition(ConversionJobTransition::Reconvert) {
661 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ConversionJobResponseDto>::error("Transition not allowed from current state")));
662 }
663
664 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
666 fields.insert("status".to_string(), serde_json::Value::String("pending".to_string()));
667
668 match backbone_core::write_guard::as_transition("reconvert", service.partial_update(&id, fields)).await {
669 Ok(Some(updated)) => {
670 let response: ConversionJobResponseDto = updated.into();
671 (StatusCode::OK, Json(ApiResponse::ok(response)))
672 }
673 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ConversionJobResponseDto>::not_found("ConversionJob", &id))),
674 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ConversionJobResponseDto>::error(e.to_string()))),
675 }
676}
677
678pub fn create_conversion_job_transition_routes(service: Arc<ConversionJobService>) -> Router {
680 use axum::routing::post;
681
682 Router::new()
683 .route("/conversion_jobs/:id/transitions/start", post(start_transition))
684 .route("/conversion_jobs/:id/transitions/update_progress", post(update_progress_transition))
685 .route("/conversion_jobs/:id/transitions/complete", post(complete_transition))
686 .route("/conversion_jobs/:id/transitions/fail", post(fail_transition))
687 .route("/conversion_jobs/:id/transitions/cancel_pending", post(cancel_pending_transition))
688 .route("/conversion_jobs/:id/transitions/cancel_running", post(cancel_running_transition))
689 .route("/conversion_jobs/:id/transitions/retry", post(retry_transition))
690 .route("/conversion_jobs/:id/transitions/reconvert", post(reconvert_transition))
691 .with_state(service)
692}
693