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::{ProcessingJobService, ServiceError};
27
28use crate::presentation::dto::{CreateProcessingJobDto, UpdateProcessingJobDto, PatchProcessingJobDto, ProcessingJobResponseDto};
30
31use crate::domain::state_machine::{ProcessingJobState, ProcessingJobStateMachine, ProcessingJobTransition};
32
33#[derive(Debug, thiserror::Error)]
35pub enum ProcessingJobError {
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("File must exist: {0}")]
46 FileNotFound(String),
47 #[error("File must be accessible: {0}")]
48 FileNotAccessible(String),
49 #[error("Job type must be supported: {0}")]
50 InvalidJobType(String),
51 #[error("Job type must match file type: {0}")]
52 JobTypeMismatch(String),
53 #[error("A similar job is already pending: {0}")]
54 DuplicateJob(String),
55 #[error("Priority must be 0-10: {0}")]
56 InvalidPriority(String),
57 #[error("Cannot modify completed jobs: {0}")]
58 JobCompleted(String),
59}
60
61impl From<ServiceError> for ProcessingJobError {
62 fn from(err: ServiceError) -> Self {
63 match err {
64 ServiceError::NotFound => Self::NotFound(err.to_string()),
65 ServiceError::Validation(ref msg) => Self::Validation(msg.clone()),
66 ServiceError::AlreadyExists(ref msg) => Self::Validation(msg.clone()),
67 ServiceError::Repository(ref e) => Self::Database(e.to_string()),
68 ServiceError::Internal(ref msg) => Self::Internal(msg.clone()),
69 ServiceError::Violations(_) => Self::Validation(err.to_string()),
70 }
71 }
72}
73
74impl axum::response::IntoResponse for ProcessingJobError {
75 fn into_response(self) -> axum::response::Response {
76 use axum::http::StatusCode;
77 use axum::Json;
78
79 let (status, code) = match &self {
80 Self::NotFound(_) => (StatusCode::NOT_FOUND, "PROCESSINGJOB_NOT_FOUND"),
81 Self::Validation(_) => (StatusCode::BAD_REQUEST, "PROCESSINGJOB_VALIDATION_ERROR"),
82 Self::Database(_) => (StatusCode::INTERNAL_SERVER_ERROR, "PROCESSINGJOB_DATABASE_ERROR"),
83 Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "PROCESSINGJOB_INTERNAL_ERROR"),
84 Self::FileNotFound(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_FILE_NOT_FOUND"),
85 Self::FileNotAccessible(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_FILE_NOT_ACCESSIBLE"),
86 Self::InvalidJobType(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_INVALID_JOB_TYPE"),
87 Self::JobTypeMismatch(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_JOB_TYPE_MISMATCH"),
88 Self::DuplicateJob(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_DUPLICATE_JOB"),
89 Self::InvalidPriority(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_INVALID_PRIORITY"),
90 Self::JobCompleted(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_JOB_COMPLETED"),
91 };
92
93 let body = serde_json::json!({
94 "success": false,
95 "error": code,
96 "message": self.to_string(),
97 });
98
99 (status, Json(body)).into_response()
100 }
101}
102
103pub mod processing_job_errors {
105 pub const FILE_NOT_FOUND: &str = "PROCESSINGJOB_FILE_NOT_FOUND";
106 pub const FILE_NOT_ACCESSIBLE: &str = "PROCESSINGJOB_FILE_NOT_ACCESSIBLE";
107 pub const INVALID_JOB_TYPE: &str = "PROCESSINGJOB_INVALID_JOB_TYPE";
108 pub const JOB_TYPE_MISMATCH: &str = "PROCESSINGJOB_JOB_TYPE_MISMATCH";
109 pub const DUPLICATE_JOB: &str = "PROCESSINGJOB_DUPLICATE_JOB";
110 pub const INVALID_PRIORITY: &str = "PROCESSINGJOB_INVALID_PRIORITY";
111 pub const JOB_COMPLETED: &str = "PROCESSINGJOB_JOB_COMPLETED";
112}
113
114pub fn create_processing_job_routes(service: Arc<ProcessingJobService>) -> Router {
147 BackboneCrudHandler::<ProcessingJobService, ProcessingJob, CreateProcessingJobDto, UpdateProcessingJobDto, ProcessingJobResponseDto>::routes(
148 service,
149 "/processing_jobs",
150 )
151}
152
153pub fn create_processing_job_read_routes(service: Arc<ProcessingJobService>) -> Router {
159 BackboneCrudHandler::<ProcessingJobService, ProcessingJob, CreateProcessingJobDto, UpdateProcessingJobDto, ProcessingJobResponseDto>::read_routes(
160 service,
161 "/processing_jobs",
162 )
163}
164
165pub fn create_processing_job_write_routes(service: Arc<ProcessingJobService>) -> Router {
177 BackboneCrudHandler::<ProcessingJobService, ProcessingJob, CreateProcessingJobDto, UpdateProcessingJobDto, ProcessingJobResponseDto>::write_routes(
178 service,
179 "/processing_jobs",
180 )
181}
182
183#[cfg(feature = "auth")]
189pub fn create_protected_processing_job_routes<A: AuthMiddleware + Send + Sync + 'static>(
190 service: Arc<ProcessingJobService>,
191 auth: Arc<A>,
192) -> Router {
193 use axum::middleware;
194 use axum::response::IntoResponse;
195
196 let auth_layer = auth.clone();
197 create_processing_job_routes(service)
198 .layer(middleware::from_fn(move |mut req: axum::extract::Request, next: axum::middleware::Next| {
199 let auth = auth_layer.clone();
200 async move {
201 let token = req.headers()
202 .get(axum::http::header::AUTHORIZATION)
203 .and_then(|h| h.to_str().ok())
204 .and_then(|raw| raw.strip_prefix("Bearer ").or_else(|| raw.strip_prefix("bearer ")))
205 .unwrap_or("");
206 match auth.authenticate(token).await {
207 Ok(ctx) => {
208 req.extensions_mut().insert(ctx);
209 next.run(req).await
210 }
211 Err(_) => {
212 (axum::http::StatusCode::UNAUTHORIZED,
213 axum::Json(serde_json::json!({
214 "success": false,
215 "error": "unauthorized",
216 "message": "Authentication required"
217 }))
218 ).into_response()
219 }
220 }
221 }
222 }))
223}
224
225pub async fn start_transition(
233 axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
234 axum::extract::Path(id): axum::extract::Path<String>,
235 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
236) -> impl axum::response::IntoResponse {
237 use axum::{http::StatusCode, Json};
238
239 let entity = match service.get_by_id(&id).await {
241 Ok(Some(e)) => e,
242 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
243 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
244 };
245
246 #[cfg(feature = "auth")]
248 {
249 let allowed_roles = ProcessingJobTransition::Start.allowed_roles();
250 let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:start");
251 let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
252 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
253 if !has_specific_perm && !has_role {
254 if !has_update_perm {
255 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for start transition")));
256 }
257 tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "start", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:start or a role of ["system"]");
258 }
259 }
260
261 let current_state: ProcessingJobState = match entity.status.to_string().parse() {
263 Ok(s) => s,
264 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
265 };
266 let sm = ProcessingJobStateMachine::from_state(current_state);
267 if !sm.can_transition(ProcessingJobTransition::Start) {
268 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
269 }
270
271 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
274 fields.insert("status".to_string(), serde_json::Value::String("running".to_string()));
275
276 match backbone_core::write_guard::as_transition("start", service.partial_update(&id, fields)).await {
277 Ok(Some(updated)) => {
278 let response: ProcessingJobResponseDto = updated.into();
279 (StatusCode::OK, Json(ApiResponse::ok(response)))
280 }
281 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
282 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
283 }
284}
285
286pub async fn complete_transition(
290 axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
291 axum::extract::Path(id): axum::extract::Path<String>,
292 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
293) -> impl axum::response::IntoResponse {
294 use axum::{http::StatusCode, Json};
295
296 let entity = match service.get_by_id(&id).await {
298 Ok(Some(e)) => e,
299 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
300 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
301 };
302
303 #[cfg(feature = "auth")]
305 {
306 let allowed_roles = ProcessingJobTransition::Complete.allowed_roles();
307 let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:complete");
308 let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
309 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
310 if !has_specific_perm && !has_role {
311 if !has_update_perm {
312 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for complete transition")));
313 }
314 tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "complete", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:complete or a role of ["system"]");
315 }
316 }
317
318 let current_state: ProcessingJobState = match entity.status.to_string().parse() {
320 Ok(s) => s,
321 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
322 };
323 let sm = ProcessingJobStateMachine::from_state(current_state);
324 if !sm.can_transition(ProcessingJobTransition::Complete) {
325 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
326 }
327
328 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
330 fields.insert("status".to_string(), serde_json::Value::String("completed".to_string()));
331
332 match backbone_core::write_guard::as_transition("complete", service.partial_update(&id, fields)).await {
333 Ok(Some(updated)) => {
334 let response: ProcessingJobResponseDto = updated.into();
335 (StatusCode::OK, Json(ApiResponse::ok(response)))
336 }
337 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
338 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
339 }
340}
341
342pub async fn fail_transition(
346 axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
347 axum::extract::Path(id): axum::extract::Path<String>,
348 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
349) -> impl axum::response::IntoResponse {
350 use axum::{http::StatusCode, Json};
351
352 let entity = match service.get_by_id(&id).await {
354 Ok(Some(e)) => e,
355 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
356 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
357 };
358
359 #[cfg(feature = "auth")]
361 {
362 let allowed_roles = ProcessingJobTransition::Fail.allowed_roles();
363 let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:fail");
364 let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
365 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
366 if !has_specific_perm && !has_role {
367 if !has_update_perm {
368 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for fail transition")));
369 }
370 tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "fail", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:fail or a role of ["system"]");
371 }
372 }
373
374 let current_state: ProcessingJobState = match entity.status.to_string().parse() {
376 Ok(s) => s,
377 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
378 };
379 let sm = ProcessingJobStateMachine::from_state(current_state);
380 if !sm.can_transition(ProcessingJobTransition::Fail) {
381 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
382 }
383
384 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
386 fields.insert("status".to_string(), serde_json::Value::String("failed".to_string()));
387
388 match backbone_core::write_guard::as_transition("fail", service.partial_update(&id, fields)).await {
389 Ok(Some(updated)) => {
390 let response: ProcessingJobResponseDto = updated.into();
391 (StatusCode::OK, Json(ApiResponse::ok(response)))
392 }
393 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
394 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
395 }
396}
397
398pub async fn cancel_pending_transition(
402 axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
403 axum::extract::Path(id): axum::extract::Path<String>,
404 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
405) -> impl axum::response::IntoResponse {
406 use axum::{http::StatusCode, Json};
407
408 let entity = match service.get_by_id(&id).await {
410 Ok(Some(e)) => e,
411 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
412 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
413 };
414
415 #[cfg(feature = "auth")]
417 {
418 let allowed_roles = ProcessingJobTransition::CancelPending.allowed_roles();
419 let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:cancel_pending");
420 let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
421 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
422 if !has_specific_perm && !has_role {
423 if !has_update_perm {
424 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for cancel_pending transition")));
425 }
426 tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "cancel_pending", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:cancel_pending or a role of ["owner", "admin"]");
427 }
428 }
429
430 let current_state: ProcessingJobState = match entity.status.to_string().parse() {
432 Ok(s) => s,
433 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
434 };
435 let sm = ProcessingJobStateMachine::from_state(current_state);
436 if !sm.can_transition(ProcessingJobTransition::CancelPending) {
437 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
438 }
439
440 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
443 fields.insert("status".to_string(), serde_json::Value::String("cancelled".to_string()));
444
445 match backbone_core::write_guard::as_transition("cancel_pending", service.partial_update(&id, fields)).await {
446 Ok(Some(updated)) => {
447 let response: ProcessingJobResponseDto = updated.into();
448 (StatusCode::OK, Json(ApiResponse::ok(response)))
449 }
450 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
451 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
452 }
453}
454
455pub async fn cancel_running_transition(
459 axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
460 axum::extract::Path(id): axum::extract::Path<String>,
461 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
462) -> impl axum::response::IntoResponse {
463 use axum::{http::StatusCode, Json};
464
465 let entity = match service.get_by_id(&id).await {
467 Ok(Some(e)) => e,
468 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
469 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
470 };
471
472 #[cfg(feature = "auth")]
474 {
475 let allowed_roles = ProcessingJobTransition::CancelRunning.allowed_roles();
476 let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:cancel_running");
477 let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
478 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
479 if !has_specific_perm && !has_role {
480 if !has_update_perm {
481 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for cancel_running transition")));
482 }
483 tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "cancel_running", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:cancel_running or a role of ["owner", "admin"]");
484 }
485 }
486
487 let current_state: ProcessingJobState = match entity.status.to_string().parse() {
489 Ok(s) => s,
490 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
491 };
492 let sm = ProcessingJobStateMachine::from_state(current_state);
493 if !sm.can_transition(ProcessingJobTransition::CancelRunning) {
494 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
495 }
496
497 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
500 fields.insert("status".to_string(), serde_json::Value::String("cancelled".to_string()));
501
502 match backbone_core::write_guard::as_transition("cancel_running", service.partial_update(&id, fields)).await {
503 Ok(Some(updated)) => {
504 let response: ProcessingJobResponseDto = updated.into();
505 (StatusCode::OK, Json(ApiResponse::ok(response)))
506 }
507 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
508 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
509 }
510}
511
512pub async fn retry_transition(
516 axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
517 axum::extract::Path(id): axum::extract::Path<String>,
518 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
519) -> impl axum::response::IntoResponse {
520 use axum::{http::StatusCode, Json};
521
522 let entity = match service.get_by_id(&id).await {
524 Ok(Some(e)) => e,
525 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
526 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
527 };
528
529 #[cfg(feature = "auth")]
531 {
532 let allowed_roles = ProcessingJobTransition::Retry.allowed_roles();
533 let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:retry");
534 let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
535 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
536 if !has_specific_perm && !has_role {
537 if !has_update_perm {
538 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for retry transition")));
539 }
540 tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "retry", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:retry or a role of ["system", "owner"]");
541 }
542 }
543
544 let current_state: ProcessingJobState = match entity.status.to_string().parse() {
546 Ok(s) => s,
547 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
548 };
549 let sm = ProcessingJobStateMachine::from_state(current_state);
550 if !sm.can_transition(ProcessingJobTransition::Retry) {
551 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
552 }
553
554 {
555 use backbone_core::row_rule as r;
556 let row = &serde_json::to_value(&entity).unwrap_or_default();
557 let state = row.get("status").and_then(|v| v.as_str()).unwrap_or_default();
558 if ["failed"].contains(&state) && !r::holds(&r::lt(r::field(row, "retry_count"), r::field(row, "max_retries"))) {
560 return (StatusCode::UNPROCESSABLE_ENTITY, Json(ApiResponse::<ProcessingJobResponseDto>::error("The retry transition's guard does not hold: retry_count < max_retries")));
561 }
562 }
563
564 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
566 fields.insert("status".to_string(), serde_json::Value::String("pending".to_string()));
567
568 match backbone_core::write_guard::as_transition("retry", service.partial_update(&id, fields)).await {
569 Ok(Some(updated)) => {
570 let response: ProcessingJobResponseDto = updated.into();
571 (StatusCode::OK, Json(ApiResponse::ok(response)))
572 }
573 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
574 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
575 }
576}
577
578pub async fn reprocess_transition(
582 axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
583 axum::extract::Path(id): axum::extract::Path<String>,
584 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
585) -> impl axum::response::IntoResponse {
586 use axum::{http::StatusCode, Json};
587
588 let entity = match service.get_by_id(&id).await {
590 Ok(Some(e)) => e,
591 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
592 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
593 };
594
595 #[cfg(feature = "auth")]
597 {
598 let allowed_roles = ProcessingJobTransition::Reprocess.allowed_roles();
599 let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:reprocess");
600 let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
601 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
602 if !has_specific_perm && !has_role {
603 if !has_update_perm {
604 return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for reprocess transition")));
605 }
606 tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "reprocess", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:reprocess or a role of ["owner", "admin"]");
607 }
608 }
609
610 let current_state: ProcessingJobState = match entity.status.to_string().parse() {
612 Ok(s) => s,
613 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
614 };
615 let sm = ProcessingJobStateMachine::from_state(current_state);
616 if !sm.can_transition(ProcessingJobTransition::Reprocess) {
617 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
618 }
619
620 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
622 fields.insert("status".to_string(), serde_json::Value::String("pending".to_string()));
623
624 match backbone_core::write_guard::as_transition("reprocess", service.partial_update(&id, fields)).await {
625 Ok(Some(updated)) => {
626 let response: ProcessingJobResponseDto = updated.into();
627 (StatusCode::OK, Json(ApiResponse::ok(response)))
628 }
629 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
630 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
631 }
632}
633
634pub fn create_processing_job_transition_routes(service: Arc<ProcessingJobService>) -> Router {
636 use axum::routing::post;
637
638 Router::new()
639 .route("/processing_jobs/:id/transitions/start", post(start_transition))
640 .route("/processing_jobs/:id/transitions/complete", post(complete_transition))
641 .route("/processing_jobs/:id/transitions/fail", post(fail_transition))
642 .route("/processing_jobs/:id/transitions/cancel_pending", post(cancel_pending_transition))
643 .route("/processing_jobs/:id/transitions/cancel_running", post(cancel_running_transition))
644 .route("/processing_jobs/:id/transitions/retry", post(retry_transition))
645 .route("/processing_jobs/:id/transitions/reprocess", post(reprocess_transition))
646 .with_state(service)
647}
648